feat: 支持推流首帧截图

api/api.go

@@ -33,6 +33,7 @@ type InviteParams struct {
 
 type StreamParams struct {
 	Stream     common.StreamID `json:"stream"`      // Source
+	Session    string          `json:"session"`     // 本次推拉流的会话ID
 	Protocol   int             `json:"protocol"`    // 推拉流协议
 	RemoteAddr string          `json:"remote_addr"` // peer地址
 }
@@ -74,6 +75,12 @@ type RecordParams struct {
 	Path string `json:"path"`
 }
 
+type SnapshotParams struct {
+	StreamParams
+	Codec        string `json:"codec"`
+	KeyFrameData []byte `json:"key_frame_data"`
+}
+
 type StreamIDParams struct {
 	StreamID common.StreamID `json:"streamid"`
 	Command  string          `json:"command"`
@@ -261,6 +268,7 @@ func StartApiServer(addr string) {
 	apiServer.router.HandleFunc("/api/v1/hook/on_receive_timeout", common.WithJsonParams(apiServer.OnReceiveTimeout, &StreamParams{}))
 	apiServer.router.HandleFunc("/api/v1/hook/on_record", common.WithJsonParams(apiServer.OnRecord, &RecordParams{}))
 	apiServer.router.HandleFunc("/api/v1/hook/on_started", apiServer.OnStarted)
+	apiServer.router.HandleFunc("/api/v1/hook/on_snapshot", common.WithJsonParams(apiServer.OnSnapshot, &SnapshotParams{}))
 
 	apiServer.registerStatisticsHandler("开始预览", "/api/v1/stream/start", withVerify(common.WithFormDataParams(apiServer.OnStreamStart, InviteParams{})))           // 实时预览
 	apiServer.registerStatisticsHandler("停止预览", "/api/v1/stream/stop", withVerify(common.WithFormDataParams(apiServer.OnCloseLiveStream, InviteParams{})))        // 关闭实时预览
@@ -315,6 +323,7 @@ func StartApiServer(addr string) {
 	apiServer.router.HandleFunc("/api/v1/getrequestkey", withVerify(func(w http.ResponseWriter, req *http.Request) {}))
 	apiServer.router.HandleFunc("/api/v1/device/positionlog", withVerify(func(w http.ResponseWriter, req *http.Request) {}))
 	apiServer.router.HandleFunc("/api/v1/device/streamlog", withVerify(func(w http.ResponseWriter, req *http.Request) {}))
+	apiServer.router.HandleFunc("/api/v1/record/list", withVerify(func(w http.ResponseWriter, req *http.Request) {}))
 
 	apiServer.registerStatisticsHandler("开始录制", "/api/v1/record/start", withVerify(apiServer.OnRecordStart)) // 开启录制
 	apiServer.registerStatisticsHandler("结束录制", "/api/v1/record/stop", withVerify(apiServer.OnRecordStop))   // 关闭录制
@@ -401,6 +410,9 @@ func StartApiServer(addr string) {
 		})
 	})
 
+	// 映射snapshot目录
+	apiServer.router.PathPrefix("/snapshot/").Handler(http.StripPrefix("/snapshot/", http.FileServer(http.Dir("./snapshot"))))
+
 	// 前端路由
 	htmlRoot := "./html/"
 	fileServer := http.FileServer(http.Dir(htmlRoot))

api/api_device.go

@@ -326,7 +326,12 @@ func (api *ApiServer) OnDeviceTree(q *QueryDeviceChannel, _ http.ResponseWriter,
 				onlineCount, _ = dao.Channel.QueryOnlineSubChannelCount(channel.RootID, channel.DeviceID, false)
 			}
 
-			response = append(response, &LiveGBSDeviceTree{Code: channel.DeviceID, Custom: false, CustomID: "", CustomName: "", ID: id, Latitude: latitude, Longitude: longitude, Manufacturer: channel.Manufacturer, Name: channel.Name, OnlineSubCount: onlineCount, Parental: false, PtzType: 0, Serial: channel.RootID, Status: channel.Status.String(), SubCount: channel.SubCount, SubCountDevice: deviceCount})
+			var ptzType int
+			if channel.Info != nil {
+				ptzType, _ = strconv.Atoi(channel.Info.PTZType)
+			}
+
+			response = append(response, &LiveGBSDeviceTree{Code: channel.DeviceID, Custom: false, CustomID: "", CustomName: "", ID: id, Latitude: latitude, Longitude: longitude, Manufacturer: channel.Manufacturer, Name: channel.Name, OnlineSubCount: onlineCount, Parental: false, PtzType: ptzType, Serial: channel.RootID, Status: channel.Status.String(), SubCount: channel.SubCount, SubCountDevice: deviceCount})
 		}
 	}
 

api/api_hook.go

@@ -6,11 +6,16 @@ import (
 	"gb-cms/hook"
 	"gb-cms/log"
 	"gb-cms/stack"
+	"github.com/csnewman/ffmpeg-go"
 	"github.com/lkmio/avformat/utils"
 	"net/http"
+	"os"
+	"path"
 	"strings"
 )
 
+var VideoKeyFrame2JPG func(codecId ffmpeg.AVCodecID, h264Data []byte, dstPath string) error
+
 func (api *ApiServer) OnPlay(params *PlayDoneParams, w http.ResponseWriter, r *http.Request) {
 	log.Sugar.Infof("播放事件. protocol: %s stream: %s", params.Protocol, params.Stream)
 
@@ -154,6 +159,7 @@ func (api *ApiServer) OnPublish(params *StreamParams, w http.ResponseWriter, _ *
 
 	stream := stack.EarlyDialogs.Find(string(params.Stream))
 	if stream != nil {
+		stream.Data = params.Session
 		stream.Put(200)
 	} else {
 		log.Sugar.Infof("推流事件. 未找到stream. stream: %s", params.Stream)
@@ -227,3 +233,44 @@ func (api *ApiServer) OnStarted(_ http.ResponseWriter, _ *http.Request) {
 		(&stack.Sink{SinkModel: sink}).Close(true, false)
 	}
 }
+
+func (api *ApiServer) OnSnapshot(v *SnapshotParams, writer http.ResponseWriter, request *http.Request) {
+	if VideoKeyFrame2JPG == nil {
+		return
+	}
+
+	var codecId ffmpeg.AVCodecID
+	switch strings.ToLower(v.Codec) {
+	case "h264":
+		codecId = ffmpeg.AVCodecIdH264
+	case "h265":
+		codecId = ffmpeg.AVCodecIdH265
+	default:
+		log.Sugar.Errorf("不支持的视频编码格式 codec: %s", v.Codec)
+		return
+	}
+
+	jpgPath := GetSnapshotPath(v.Stream, v.Session)
+	err := os.MkdirAll(path.Dir(jpgPath), 0755)
+	if err != nil {
+		log.Sugar.Errorf("创建目录失败 err: %s", err.Error())
+		return
+	}
+
+	// 关家帧转换并保存JPEG
+	err = VideoKeyFrame2JPG(codecId, v.KeyFrameData, jpgPath)
+	if err != nil {
+		log.Sugar.Errorf("转换为JPEG失败 err: %s", err.Error())
+	} else if err = dao.Channel.SetSnapshotPath(v.Stream.DeviceID(), v.Stream.ChannelID(), jpgPath); err != nil {
+		// 数据库更新通道的最新截图
+		log.Sugar.Errorf("更新通道最新截图失败 err: %s", err.Error())
+	}
+}
+
+func GetSnapshotPath(streamID common.StreamID, sessionID string) string {
+	if VideoKeyFrame2JPG == nil {
+		return ""
+	}
+
+	return path.Join("./snapshot/", string(streamID), sessionID+".jpg")
+}

api/api_stream.go

@@ -95,7 +95,7 @@ func (api *ApiServer) DoStartStream(v *InviteParams, w http.ResponseWriter, r *h
 	}
 
 	response := LiveGBSStream{
-		AudioEnable:           false,
+		AudioEnable:           true,
 		CDN:                   "",
 		CascadeSize:           0,
 		ChannelID:             v.ChannelID,
@@ -116,7 +116,7 @@ func (api *ApiServer) DoStartStream(v *InviteParams, w http.ResponseWriter, r *h
 		RecordStartAt:         "",
 		RelaySize:             0,
 		SMSID:                 "",
-		SnapURL:               "",
+		SnapURL:               GetSnapshotPath(stream.StreamID, stream.SessionID),
 		SourceAudioCodecName:  "",
 		SourceAudioSampleRate: 0,
 		SourceVideoCodecName:  "",

api/api_system.go

@@ -148,7 +148,7 @@ func (api *ApiServer) OnSetBaseConfig(baseConfig *BaseConfig, _ http.ResponseWri
 	}
 
 	// 更新优先流格式
-	if baseConfig.PreferStreamFmt != "" && baseConfig.PreferStreamFmt != common.Config.PreferStreamFmt {
+	if baseConfig.PreferStreamFmt != common.Config.PreferStreamFmt {
 		iniConfig.Section("sip").Key("prefer_stream_fmt").SetValue(baseConfig.PreferStreamFmt)
 		changed = true
 	}

api/livegbs_bean.go

@@ -307,7 +307,7 @@ func ChannelModels2LiveGBSChannels(index int, channels []*dao.ChannelModel, devi
 			SerialNumber:       "",
 			Shared:             false,
 			SignalLevel:        0,
-			SnapURL:            "",
+			SnapURL:            channel.SnapshotPath,
 			Speed:              0,
 			Status:             channel.Status.String(),
 			StreamID:           string(streamID), // 实时流ID

api/snapshot.go

@@ -0,0 +1,120 @@
+//go:build (darwin && (amd64 || arm64)) || (linux && (amd64 || arm64)) || (amd64 && windows)
+
+package api
+
+import (
+	"fmt"
+	"github.com/csnewman/ffmpeg-go"
+	"os"
+	"unsafe"
+)
+
+func init() {
+	VideoKeyFrame2JPG = videoKeyFrame2JPG
+}
+
+func videoKeyFrame2JPG(codecId ffmpeg.AVCodecID, h264Data []byte, dstPath string) error {
+	// 2. 创建解码器
+	codec := ffmpeg.AVCodecFindDecoder(codecId)
+	if codec == nil {
+		return fmt.Errorf("找不到解码器 %v", codecId)
+	}
+
+	codecCtx := ffmpeg.AVCodecAllocContext3(codec)
+	defer ffmpeg.AVCodecFreeContext(&codecCtx)
+
+	// 3. 打开解码器
+	if _, err := ffmpeg.AVCodecOpen2(codecCtx, codec, nil); err != nil {
+		return fmt.Errorf("打开解码器失败 %v", err)
+	}
+
+	// 4. 创建AVPacket并填充数据
+	pkt := ffmpeg.AVPacketAlloc()
+	defer ffmpeg.AVPacketFree(&pkt)
+
+	// 将H264数据拷贝到AVPacket
+	pktBuf := ffmpeg.AVMalloc(uint64(len(h264Data)))
+	copy(unsafe.Slice((*byte)(pktBuf), len(h264Data)), h264Data)
+	pkt.SetData(pktBuf)
+	pkt.SetSize(len(h264Data))
+	pkt.SetFlags(pkt.Flags() | ffmpeg.AVPktFlagKey) // 标记为关键帧
+
+	// 5. 解码
+	frame := ffmpeg.AVFrameAlloc()
+	defer ffmpeg.AVFrameFree(&frame)
+
+	// 发送数据包到解码器
+	if _, err := ffmpeg.AVCodecSendPacket(codecCtx, pkt); err != nil {
+		return fmt.Errorf("发送包失败 %v", err)
+	}
+
+	// 接收解码后的帧
+	if _, err := ffmpeg.AVCodecReceiveFrame(codecCtx, frame); err != nil {
+		return fmt.Errorf("解码失败 %v", err)
+	}
+
+	// 6. 保存为JPEG
+	err := saveFrameAsJPEG(frame, dstPath)
+	return err
+}
+
+func saveFrameAsJPEG(frame *ffmpeg.AVFrame, filename string) error {
+	// 创建JPEG编码器
+	codec := ffmpeg.AVCodecFindEncoder(ffmpeg.AVCodecIdMjpeg)
+	if codec == nil {
+		return fmt.Errorf("找不到JPEG编码器")
+	}
+
+	codecCtx := ffmpeg.AVCodecAllocContext3(codec)
+	defer ffmpeg.AVCodecFreeContext(&codecCtx)
+
+	// 设置编码参数
+	codecCtx.SetPixFmt(ffmpeg.AVPixFmtYuvj420P)
+	codecCtx.SetWidth(frame.Width())
+	codecCtx.SetHeight(frame.Height())
+	rational := ffmpeg.AVRational{}
+	rational.SetNum(1)
+	rational.SetDen(25)
+	codecCtx.SetTimeBase(&rational)
+
+	codecCtx.SetColorspace(frame.Colorspace())
+	codecCtx.SetColorRange(frame.ColorRange())
+
+	strict := ffmpeg.ToCStr("strict")
+	defer strict.Free()
+
+	if _, err := ffmpeg.AVOptSetInt(codecCtx.RawPtr(), strict, ffmpeg.FFComplianceUnofficial, 0); err != nil {
+		return fmt.Errorf("警告: 设置strict参数失败 %v", err)
+	}
+
+	// 打开编码器
+	if _, err := ffmpeg.AVCodecOpen2(codecCtx, codec, nil); err != nil {
+		return fmt.Errorf("打开JPEG编码器失败 %v", err)
+	}
+
+	// 创建输出文件
+	file, err := os.Create(filename)
+	if err != nil {
+		return fmt.Errorf("创建文件失败 %v", err)
+	}
+	defer file.Close()
+
+	// 编码帧
+	pkt := ffmpeg.AVPacketAlloc()
+	defer ffmpeg.AVPacketFree(&pkt)
+
+	if _, err := ffmpeg.AVCodecSendFrame(codecCtx, frame); err != nil {
+		return fmt.Errorf("发送帧失败 %v", err)
+	}
+
+	if _, err := ffmpeg.AVCodecReceivePacket(codecCtx, pkt); err != nil {
+		return fmt.Errorf("接收包失败 %v", err)
+	}
+
+	// 写入文件
+	if _, err := file.Write(unsafe.Slice((*byte)(pkt.Data()), pkt.Size())); err != nil {
+		return fmt.Errorf("写入文件失败 %v", err)
+	}
+
+	return nil
+}

dao/channel.go

@@ -88,6 +88,7 @@ type ChannelModel struct {
 	CustomID        *string             `gorm:"unique" xml:"-"`                   // 自定义通道ID
 	Event           string              `json:"-" xml:"Event,omitempty" gorm:"-"` // <!-- 状态改变事件ON:上线,OFF:离线,VLOST:视频丢失,DEFECT:故障,ADD:增加,DEL:删除,UPDATE:更新(必选)-->
 	DropMark        int                 `json:"-" xml:"-"`                        // 是否被过滤 0-不被过滤/非0-被过滤
+	SnapshotPath    string              `json:"snapshot_path" xml:"-"`            // 快照路径
 }
 
 func (d *ChannelModel) TableName() string {
@@ -430,3 +431,7 @@ func (d *daoChannel) DropChannel(rootId string, typeCodes []string, tx *gorm.DB)
 		return update(tx)
 	})
 }
+
+func (d *daoChannel) SetSnapshotPath(rootId string, channelId string, snapshotPath string) error {
+	return db.Model(&ChannelModel{}).Where("root_id =? and device_id =?", rootId, channelId).Update("snapshot_path", snapshotPath).Error
+}

dao/stream.go

@@ -12,6 +12,7 @@ type StreamModel struct {
 	DeviceID   string                 `gorm:"index"`                         // 下级设备ID, 统计某个设备的所有流/1078设备为sim number
 	ChannelID  string                 `gorm:"index"`                         // 下级通道ID, 统计某个设备下的某个通道的所有流/1078设备为 channel number
 	StreamID   common.StreamID        `json:"stream_id" gorm:"index,unique"` // 流ID
+	SessionID  string                 `gorm:"index"`                         // 某次推流的会话ID
 	Protocol   int                    `json:"protocol,omitempty"`            // 推流协议, @See stack.SourceTypeRtmp
 	StreamType string                 // play/playback/download
 	Dialog     *common.RequestWrapper `json:"dialog,omitempty"` // 国标流的SipCall会话

go.mod

@@ -53,6 +53,7 @@ require (
 )
 
 require (
+	github.com/csnewman/ffmpeg-go v0.6.0
 	github.com/glebarez/sqlite v1.11.0
 	github.com/go-co-op/gocron/v2 v2.16.6
 	github.com/gorilla/mux v1.8.1
@@ -66,3 +67,5 @@ require (
 )
 
 replace github.com/ghettovoice/gosip => github.com/lkmio/gosip v0.0.0-20251016021306-565c7a2fa4f5
+
+replace github.com/csnewman/ffmpeg-go => github.com/lkmio/ffmpeg-go v0.0.0-20251129021554-ca32d075a5b7

main.go

@@ -98,6 +98,10 @@ func main() {
 	log.Sugar.Infof("启动http server. addr: %s", httpAddr)
 	go api.StartApiServer(httpAddr)
 
+	if api.VideoKeyFrame2JPG != nil {
+		log.Sugar.Infof("snapshot enabled")
+	}
+
 	err = http.ListenAndServe(":19000", nil)
 	if err != nil {
 		println(err)

stack/live.go

@@ -59,13 +59,17 @@ func (d *Device) StartStream(inviteType common.InviteType, streamId common.Strea
 		if !ok {
 			log.Sugar.Infof("收流超时 发送bye请求...")
 			CloseStream(streamId, true)
+		} else if waiting.Data != nil {
+			if session, ok := waiting.Data.(string); ok {
+				stream.SessionID = session
+			}
 		}
 		return ok
 	}
 
-	if sync {
+	if !sync {
 		go wait()
-	} else if !sync && !wait() {
+	} else if sync && !wait() {
 		return nil, fmt.Errorf("receiving stream timed out")
 	}