weijuesen
289 lines
9.1 KiB
Go
289 lines
9.1 KiB
Go
package handler
|
||
|
||
import (
|
||
"bytes"
|
||
"encoding/json"
|
||
"fmt"
|
||
"io"
|
||
"log/slog"
|
||
"net/http"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"silk-server-go/internal/config"
|
||
"silk-server-go/internal/middleware"
|
||
"silk-server-go/internal/model"
|
||
"silk-server-go/internal/service"
|
||
|
||
"github.com/gin-gonic/gin"
|
||
"gorm.io/gorm"
|
||
)
|
||
|
||
// ActiveRecording 活跃录制记录（内存维护）
|
||
type ActiveRecording struct {
|
||
CameraID string `json:"cameraId"`
|
||
DeviceID string `json:"deviceId"`
|
||
ChannelID string `json:"channelId"`
|
||
Stream string `json:"stream"`
|
||
App string `json:"app"`
|
||
StartedAt time.Time `json:"startedAt"`
|
||
}
|
||
|
||
// 活跃录制内存表（cameraId -> recording）
|
||
var (
|
||
activeRecordings = make(map[string]*ActiveRecording)
|
||
activeMu sync.Mutex
|
||
)
|
||
|
||
// RegisterVideoRecordRoutes 注册录制管理路由
|
||
func RegisterVideoRecordRoutes(rg *gin.RouterGroup, db *gorm.DB, media *service.MediaService, cfg *config.Config) {
|
||
recordPerm := middleware.RequirePermission(db, "video:record")
|
||
readPerm := middleware.RequirePermission(db, "video:read")
|
||
rg.POST("/video/cameras/:id/record/start", recordPerm, startRecording(db, media, cfg))
|
||
rg.POST("/video/cameras/:id/record/stop", recordPerm, stopRecording(db, media, cfg))
|
||
rg.GET("/video/recordings/active", readPerm, listActiveRecordings(cfg))
|
||
rg.POST("/video/recordings/internal/end", endRecordingInternal(media, cfg)) // 白名单接口，无需权限
|
||
}
|
||
|
||
// startRecording 开始录制：查摄像头 → media.StartPlay → POST recorderApiBase/record/start
|
||
func startRecording(db *gorm.DB, media *service.MediaService, cfg *config.Config) gin.HandlerFunc {
|
||
return func(c *gin.Context) {
|
||
cameraId := c.Param("id")
|
||
var camera model.Camera
|
||
if db.Where("id = ?", cameraId).First(&camera).Error != nil {
|
||
c.JSON(http.StatusNotFound, gin.H{"error": "camera not found"})
|
||
return
|
||
}
|
||
if camera.GbDeviceID == nil || camera.GbChannelID == nil ||
|
||
*camera.GbDeviceID == "" || *camera.GbChannelID == "" {
|
||
c.JSON(http.StatusBadRequest, gin.H{"error": "摄像头未配置 GB28181 或不存在"})
|
||
return
|
||
}
|
||
|
||
// 检查是否已在录制
|
||
activeMu.Lock()
|
||
if _, ok := activeRecordings[cameraId]; ok {
|
||
activeMu.Unlock()
|
||
c.JSON(http.StatusConflict, gin.H{"error": "该摄像头正在录制中"})
|
||
return
|
||
}
|
||
activeMu.Unlock()
|
||
|
||
// 1. 确保视频流在线（WVP play/start）
|
||
_, err := media.StartPlay(*camera.GbDeviceID, *camera.GbChannelID)
|
||
if err != nil {
|
||
slog.Warn("开始播放失败，跳过录制", "cameraId", cameraId, "error", err)
|
||
// StartPlay 失败，标记摄像头离线
|
||
db.Model(&model.Camera{}).Where("id = ?", cameraId).Update("is_online", false)
|
||
c.JSON(http.StatusServiceUnavailable, gin.H{"error": "摄像头可能离线: " + err.Error()})
|
||
return
|
||
}
|
||
|
||
// 2. 通知 Python 录制服务开始录制
|
||
stream := *camera.GbDeviceID + "_" + *camera.GbChannelID
|
||
recorderBase := strings.TrimSuffix(cfg.RecorderAPIBase, "/")
|
||
payload, _ := json.Marshal(map[string]string{
|
||
"stream": stream,
|
||
"cameraId": cameraId,
|
||
"deviceId": *camera.GbDeviceID,
|
||
"channelId": *camera.GbChannelID,
|
||
})
|
||
|
||
client := &http.Client{Timeout: 10 * time.Second}
|
||
resp, err := client.Post(recorderBase+"/record/start", "application/json", bytes.NewReader(payload))
|
||
if err != nil {
|
||
c.JSON(http.StatusServiceUnavailable, gin.H{"error": "录制服务不可用: " + err.Error()})
|
||
return
|
||
}
|
||
defer resp.Body.Close()
|
||
if resp.StatusCode != http.StatusOK {
|
||
respBody, _ := io.ReadAll(resp.Body)
|
||
c.JSON(http.StatusServiceUnavailable, gin.H{"error": fmt.Sprintf("录制服务返回 %d: %s", resp.StatusCode, string(respBody))})
|
||
return
|
||
}
|
||
|
||
// 3. 记录活跃录制
|
||
rec := &ActiveRecording{
|
||
CameraID: cameraId,
|
||
DeviceID: *camera.GbDeviceID,
|
||
ChannelID: *camera.GbChannelID,
|
||
Stream: stream,
|
||
App: "rtp",
|
||
StartedAt: time.Now(),
|
||
}
|
||
activeMu.Lock()
|
||
activeRecordings[cameraId] = rec
|
||
activeMu.Unlock()
|
||
|
||
slog.Info("录制已开始", "cameraId", cameraId, "stream", stream)
|
||
c.JSON(http.StatusOK, gin.H{
|
||
"cameraId": cameraId,
|
||
"stream": stream,
|
||
"startedAt": rec.StartedAt,
|
||
})
|
||
}
|
||
}
|
||
|
||
// stopRecording 停止录制：通知录制服务 + WVP play/stop 清理 session
|
||
func stopRecording(db *gorm.DB, media *service.MediaService, cfg *config.Config) gin.HandlerFunc {
|
||
return func(c *gin.Context) {
|
||
cameraId := c.Param("id")
|
||
|
||
activeMu.Lock()
|
||
rec, ok := activeRecordings[cameraId]
|
||
if ok {
|
||
delete(activeRecordings, cameraId)
|
||
}
|
||
activeMu.Unlock()
|
||
|
||
// 即使没有活跃录制，也尝试用摄像头的 GB 信息停止 WVP play
|
||
var deviceID, channelID string
|
||
if ok {
|
||
deviceID = rec.DeviceID
|
||
channelID = rec.ChannelID
|
||
} else {
|
||
var camera model.Camera
|
||
if db.Where("id = ?", cameraId).First(&camera).Error == nil &&
|
||
camera.GbDeviceID != nil && camera.GbChannelID != nil {
|
||
deviceID = *camera.GbDeviceID
|
||
channelID = *camera.GbChannelID
|
||
}
|
||
}
|
||
|
||
// 1. 通知录制服务停止
|
||
if ok {
|
||
recorderBase := strings.TrimSuffix(cfg.RecorderAPIBase, "/")
|
||
payload, _ := json.Marshal(map[string]string{
|
||
"stream": rec.Stream,
|
||
})
|
||
client := &http.Client{Timeout: 10 * time.Second}
|
||
resp, err := client.Post(recorderBase+"/record/stop", "application/json", bytes.NewReader(payload))
|
||
if err != nil {
|
||
slog.Warn("停止录制服务失败", "cameraId", cameraId, "error", err)
|
||
} else {
|
||
resp.Body.Close()
|
||
}
|
||
}
|
||
|
||
// 2. 停止 WVP play session（清理 ZLMediaKit 流）
|
||
if deviceID != "" && channelID != "" {
|
||
media.StopPlay(deviceID, channelID)
|
||
}
|
||
|
||
slog.Info("录制已停止", "cameraId", cameraId, "hadActiveRecording", ok)
|
||
c.JSON(http.StatusOK, gin.H{
|
||
"cameraId": cameraId,
|
||
"stopped": true,
|
||
})
|
||
}
|
||
}
|
||
|
||
// endRecordingInternal 录制服务内部接口：通知录制已结束（流 EOF / 错误 / 停止）
|
||
// 清理 activeRecordings 并停止 WVP play session
|
||
func endRecordingInternal(media *service.MediaService, cfg *config.Config) gin.HandlerFunc {
|
||
return func(c *gin.Context) {
|
||
apiKey := c.GetHeader("x-api-key")
|
||
if apiKey != cfg.InternalAPIKey {
|
||
c.JSON(http.StatusUnauthorized, gin.H{"error": "invalid api key"})
|
||
return
|
||
}
|
||
|
||
var body struct {
|
||
Stream string `json:"stream"`
|
||
CameraID string `json:"cameraId"`
|
||
}
|
||
if err := c.ShouldBindJSON(&body); err != nil {
|
||
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
||
return
|
||
}
|
||
|
||
activeMu.Lock()
|
||
rec, ok := activeRecordings[body.CameraID]
|
||
if ok {
|
||
delete(activeRecordings, body.CameraID)
|
||
}
|
||
activeMu.Unlock()
|
||
|
||
if ok {
|
||
slog.Info("录制服务通知录制结束", "cameraId", body.CameraID, "stream", body.Stream)
|
||
// 停止 WVP play session，清理 ZLMediaKit 流
|
||
media.StopPlay(rec.DeviceID, rec.ChannelID)
|
||
}
|
||
|
||
c.JSON(http.StatusOK, gin.H{"ok": true})
|
||
}
|
||
}
|
||
|
||
// listActiveRecordings 返回活跃录制列表
|
||
// 优先以 recorder-go 服务的实际状态为准（Go 后端重启后内存 map 会丢失，但 recorder-go 仍在录制）
|
||
func listActiveRecordings(cfg *config.Config) gin.HandlerFunc {
|
||
return func(c *gin.Context) {
|
||
// 1. 查询 recorder-go 实际录制状态
|
||
recorderBase := strings.TrimSuffix(cfg.RecorderAPIBase, "/")
|
||
client := &http.Client{Timeout: 3 * time.Second}
|
||
resp, err := client.Get(recorderBase + "/record/status")
|
||
if err == nil {
|
||
defer resp.Body.Close()
|
||
if resp.StatusCode == http.StatusOK {
|
||
var statusResp struct {
|
||
Recordings []struct {
|
||
Stream string `json:"stream"`
|
||
CameraID string `json:"cameraId"`
|
||
StartedAt string `json:"startedAt"`
|
||
TotalBytes int64 `json:"totalBytes"`
|
||
SegmentCount int `json:"segmentCount"`
|
||
} `json:"recordings"`
|
||
}
|
||
if err := json.NewDecoder(resp.Body).Decode(&statusResp); err == nil && len(statusResp.Recordings) > 0 {
|
||
list := make([]*ActiveRecording, 0, len(statusResp.Recordings))
|
||
for _, r := range statusResp.Recordings {
|
||
startedAt, _ := time.Parse(time.RFC3339, r.StartedAt)
|
||
if startedAt.IsZero() {
|
||
startedAt = time.Now()
|
||
}
|
||
// 拆分 stream 得到 deviceId/channelId
|
||
deviceID, channelID := "", ""
|
||
parts := strings.SplitN(r.Stream, "_", 2)
|
||
if len(parts) == 2 {
|
||
deviceID, channelID = parts[0], parts[1]
|
||
}
|
||
list = append(list, &ActiveRecording{
|
||
CameraID: r.CameraID,
|
||
DeviceID: deviceID,
|
||
ChannelID: channelID,
|
||
Stream: r.Stream,
|
||
App: "rtp",
|
||
StartedAt: startedAt,
|
||
})
|
||
// 同步更新内存 map（自愈：Go 后端重启后内存丢失，从 recorder-go 恢复）
|
||
activeMu.Lock()
|
||
if _, ok := activeRecordings[r.CameraID]; !ok {
|
||
activeRecordings[r.CameraID] = &ActiveRecording{
|
||
CameraID: r.CameraID,
|
||
DeviceID: deviceID,
|
||
ChannelID: channelID,
|
||
Stream: r.Stream,
|
||
App: "rtp",
|
||
StartedAt: startedAt,
|
||
}
|
||
}
|
||
activeMu.Unlock()
|
||
}
|
||
c.JSON(http.StatusOK, list)
|
||
return
|
||
}
|
||
}
|
||
}
|
||
|
||
// 2. recorder-go 不可用时回退到内存 map
|
||
activeMu.Lock()
|
||
list := make([]*ActiveRecording, 0, len(activeRecordings))
|
||
for _, rec := range activeRecordings {
|
||
list = append(list, rec)
|
||
}
|
||
activeMu.Unlock()
|
||
c.JSON(http.StatusOK, list)
|
||
}
|
||
}
|