monibuca/recoder.go

package m7s

import (
	"os"
	"path/filepath"
	"time"

	"gorm.io/gorm"

	"m7s.live/v5/pkg/config"

	"m7s.live/v5/pkg/task"

	"m7s.live/v5/pkg"
)

type (
	IRecorder interface {
		task.ITask
		GetRecordJob() *RecordJob
	}
	RecorderFactory = func(config.Record) IRecorder
	// RecordEvent 包含录像事件的公共字段

	EventRecordStream struct {
		*config.RecordEvent
		RecordStream
	}
	RecordJob struct {
		task.Job
		Event      *config.RecordEvent
		StreamPath string // 对应本地流
		Plugin     *Plugin
		Subscriber *Subscriber
		SubConf    *config.Subscribe
		RecConf    *config.Record
		recorder   IRecorder
	}
	DefaultRecorder struct {
		task.Task
		RecordJob RecordJob
		Event     EventRecordStream
	}
	RecordStream struct {
		ID         uint      `gorm:"primarykey"`
		StartTime  time.Time `gorm:"default:NULL"`
		EndTime    time.Time `gorm:"default:NULL"`
		Duration   uint32    `gorm:"comment:录像时长;default:0"`
		Filename   string    `json:"fileName" desc:"文件名" gorm:"type:varchar(255);comment:文件名"`
		Type       string    `json:"type" desc:"录像文件类型" gorm:"type:varchar(255);comment:录像文件类型,flv,mp4,raw,fmp4,hls"`
		FilePath   string
		StreamPath string
		AudioCodec string
		VideoCodec string
		CreatedAt  time.Time
		DeletedAt  gorm.DeletedAt `gorm:"index" yaml:"-"`
	}
)

func (r *DefaultRecorder) GetRecordJob() *RecordJob {
	return &r.RecordJob
}

func (r *DefaultRecorder) Start() (err error) {
	return r.RecordJob.Subscribe()
}

func (r *DefaultRecorder) CreateStream(start time.Time, customFileName func(*RecordJob) string) (err error) {
	recordJob := &r.RecordJob
	sub := recordJob.Subscriber
	r.Event.RecordStream = RecordStream{
		StartTime:  start,
		StreamPath: sub.StreamPath,
		FilePath:   customFileName(recordJob),
		Type:       recordJob.RecConf.Type,
	}
	dir := filepath.Dir(r.Event.FilePath)
	if err = os.MkdirAll(dir, 0755); err != nil {
		return
	}
	if sub.Publisher.HasAudioTrack() {
		r.Event.AudioCodec = sub.Publisher.AudioTrack.ICodecCtx.String()
	}
	if sub.Publisher.HasVideoTrack() {
		r.Event.VideoCodec = sub.Publisher.VideoTrack.ICodecCtx.String()
	}
	if recordJob.Plugin.DB != nil {
		if recordJob.Event != nil {
			r.Event.RecordEvent = recordJob.Event
			recordJob.Plugin.DB.Save(&r.Event)
		} else {
			recordJob.Plugin.DB.Save(&r.Event.RecordStream)
		}
	}
	return
}

func (r *DefaultRecorder) WriteTail(end time.Time, tailJob task.IJob) {
	r.Event.EndTime = end
	if r.RecordJob.Plugin.DB != nil {
		// 将事件和录像记录关联
		if r.RecordJob.Event != nil {
			r.RecordJob.Plugin.DB.Save(&r.Event)
		} else {
			r.RecordJob.Plugin.DB.Save(&r.Event.RecordStream)
		}
	}
	if tailJob == nil {
		return
	}
	tailJob.AddTask(NewEventRecordCheck(r.Event.Type, r.Event.StreamPath, r.RecordJob.Plugin.DB))
}

func (p *RecordJob) GetKey() string {
	return p.RecConf.FilePath
}

func (p *RecordJob) Subscribe() (err error) {

	p.Subscriber, err = p.Plugin.SubscribeWithConfig(p.recorder.GetTask().Context, p.StreamPath, *p.SubConf)
	return
}

func (p *RecordJob) Init(recorder IRecorder, plugin *Plugin, streamPath string, conf config.Record, subConf *config.Subscribe) *RecordJob {
	p.Plugin = plugin
	p.RecConf = &conf
	p.StreamPath = streamPath
	if subConf == nil {
		conf := p.Plugin.config.Subscribe
		subConf = &conf
	}
	subConf.SubType = SubscribeTypeVod
	p.SubConf = subConf
	p.recorder = recorder
	p.SetDescriptions(task.Description{
		"plugin":     plugin.Meta.Name,
		"streamPath": streamPath,
		"filePath":   conf.FilePath,
		"append":     conf.Append,
		"fragment":   conf.Fragment,
	})
	recorder.SetRetry(-1, time.Second)
	if sender := plugin.getHookSender(config.HookOnRecordStart); sender != nil {
		recorder.OnStart(func() {
			webhookData := map[string]interface{}{
				"event":      config.HookOnRecordStart,
				"streamPath": streamPath,
				"filePath":   conf.FilePath,
				"pluginName": plugin.Meta.Name,
				"timestamp":  time.Now().Unix(),
			}
			sender(config.HookOnRecordStart, webhookData)
		})
	}

	if sender := plugin.getHookSender(config.HookOnRecordEnd); sender != nil {
		recorder.OnDispose(func() {
			webhookData := map[string]interface{}{
				"event":      config.HookOnRecordEnd,
				"streamPath": streamPath,
				"filePath":   conf.FilePath,
				"reason":     recorder.StopReason().Error(),
				"timestamp":  time.Now().Unix(),
			}
			sender(config.HookOnRecordEnd, webhookData)
		})
	}

	plugin.Server.Records.Add(p, plugin.Logger.With("filePath", conf.FilePath, "streamPath", streamPath))
	return p
}

func (p *RecordJob) Start() (err error) {
	s := p.Plugin.Server
	if _, ok := s.Records.Get(p.GetKey()); ok {
		return pkg.ErrRecordSamePath
	}
	// dir := p.FilePath
	// if p.Fragment == 0 || p.Append {
	// 	dir = filepath.Dir(p.FilePath)
	// }
	// p.SetDescription("filePath", p.FilePath)
	// if err = os.MkdirAll(dir, 0755); err != nil {
	// 	return
	// }
	p.AddTask(p.recorder, p.Logger)
	return
}

func NewEventRecordCheck(t string, streamPath string, db *gorm.DB) *eventRecordCheck {
	return &eventRecordCheck{
		DB:         db,
		streamPath: streamPath,
		Type:       t,
	}
}

type eventRecordCheck struct {
	task.Task
	DB         *gorm.DB
	streamPath string
	Type       string
}

func (t *eventRecordCheck) Run() (err error) {
	var eventRecordStreams []EventRecordStream
	t.DB.Find(&eventRecordStreams, "type=? AND event_level=high AND stream_path=?", t.Type, t.streamPath) //搜索事件录像，且为重要事件（无法自动删除）
	for _, recordStream := range eventRecordStreams {
		t.DB.Model(&EventRecordStream{}).Where(`event_level=low AND start_time <= ? and end_time >= ?`, recordStream.EndTime, recordStream.StartTime).Update("event_level", config.EventLevelHigh)
	}
	return
}