feat: 自动建立实时监看媒体路径 (#51)

This commit is contained in:
QiuSW
2026-08-13 12:39:12 +08:00
parent 73d775b199
commit b9f92dd46e
12 changed files with 181 additions and 18 deletions
@@ -23,3 +23,8 @@ CREATE TABLE IF NOT EXISTS sense_admission_profiles (
PRIMARY KEY(device_id, token)
);
`
const MediaStatusMigrationSQL = `
ALTER TABLE sense_admission_results ADD COLUMN IF NOT EXISTS media_status TEXT NOT NULL DEFAULT '';
ALTER TABLE sense_admission_results ADD COLUMN IF NOT EXISTS media_detail TEXT NOT NULL DEFAULT '';
`
+27 -6
View File
@@ -24,22 +24,35 @@ type Profile struct {
Verification rtsp.Result `json:"verification"`
}
type Result struct {
DeviceID string `json:"device_id"`
Address string `json:"address"`
Status string `json:"status"`
Detail string `json:"detail"`
Profiles []Profile `json:"profiles"`
CheckedAt time.Time `json:"checked_at"`
DeviceID string `json:"device_id"`
Address string `json:"address"`
Status string `json:"status"`
Detail string `json:"detail"`
Profiles []Profile `json:"profiles"`
MediaStatus string `json:"media_status,omitempty"`
MediaDetail string `json:"media_detail,omitempty"`
CheckedAt time.Time `json:"checked_at"`
}
type MediaOutcome struct {
Status string
Detail string
}
type ReadyHandler func(context.Context, identity.Principal, Result) MediaOutcome
type Service struct {
onvif onvif.Client
rtsp rtsp.Verifier
discoveryIP string
discoveryTimeout time.Duration
store Store
readyHandler ReadyHandler
now func() time.Time
}
func (s *Service) SetReadyHandler(handler ReadyHandler) { s.readyHandler = handler }
func NewService(client onvif.Client, verifier rtsp.Verifier, discoveryIP string, stores ...Store) *Service {
var store Store = NewMemoryStore()
if len(stores) > 0 && stores[0] != nil {
@@ -106,6 +119,14 @@ func (s *Service) Probe(ctx context.Context, actor identity.Principal, deviceID,
break
}
}
if s.readyHandler != nil {
outcome := s.readyHandler(ctx, actor, result)
result.MediaStatus = outcome.Status
result.MediaDetail = outcome.Detail
if err := s.store.Save(ctx, result); err != nil {
return Result{}, err
}
}
identity.RecordAudit(ctx, actor.UserID, "admission.probe", deviceID, "success", map[string]any{"profile_count": len(items), "status": status})
return result, nil
}
@@ -78,6 +78,12 @@ func TestProbeUsesSeparateCredentialsPersistsProfilesAndActivatesDevice(t *testi
rtspVerifier := &credentialCapturingRTSP{}
store := NewMemoryStore()
service := NewService(onvifClient, rtspVerifier, "", store)
service.SetReadyHandler(func(_ context.Context, _ identity.Principal, result Result) MediaOutcome {
if len(result.Profiles) != 2 {
t.Fatalf("ready handler profiles=%d", len(result.Profiles))
}
return MediaOutcome{Status: "needs_attention", Detail: "媒体服务未就绪"}
})
result, err := service.Probe(context.Background(), identity.Principal{}, item.ID, "http://camera.invalid/onvif/device_service")
if err != nil {
t.Fatal(err)
@@ -85,6 +91,9 @@ func TestProbeUsesSeparateCredentialsPersistsProfilesAndActivatesDevice(t *testi
if onvifClient.got.Username != "onvif-user" || rtspVerifier.got.Username != "rtsp-user" {
t.Fatal("credentials were not separated")
}
if result.MediaStatus != "needs_attention" {
t.Fatalf("media outcome=%#v", result)
}
restarted := NewService(onvifClient, rtspVerifier, "", store)
persisted, ok := restarted.Get(item.ID)
if !ok || len(persisted.Profiles) != len(result.Profiles) {
+2 -2
View File
@@ -45,7 +45,7 @@ func (s *PostgresStore) Save(ctx context.Context, result Result) error {
return err
}
defer tx.Rollback()
_, err = tx.ExecContext(ctx, `INSERT INTO sense_admission_results(device_id,address,status,detail,checked_at) VALUES($1,$2,$3,$4,$5) ON CONFLICT(device_id) DO UPDATE SET address=EXCLUDED.address,status=EXCLUDED.status,detail=EXCLUDED.detail,checked_at=EXCLUDED.checked_at`, result.DeviceID, result.Address, result.Status, result.Detail, result.CheckedAt)
_, err = tx.ExecContext(ctx, `INSERT INTO sense_admission_results(device_id,address,status,detail,media_status,media_detail,checked_at) VALUES($1,$2,$3,$4,$5,$6,$7) ON CONFLICT(device_id) DO UPDATE SET address=EXCLUDED.address,status=EXCLUDED.status,detail=EXCLUDED.detail,media_status=EXCLUDED.media_status,media_detail=EXCLUDED.media_detail,checked_at=EXCLUDED.checked_at`, result.DeviceID, result.Address, result.Status, result.Detail, result.MediaStatus, result.MediaDetail, result.CheckedAt)
if err != nil {
return err
}
@@ -62,7 +62,7 @@ func (s *PostgresStore) Save(ctx context.Context, result Result) error {
}
func (s *PostgresStore) Get(ctx context.Context, deviceID string) (Result, error) {
var result Result
err := s.database.QueryRowContext(ctx, `SELECT device_id,address,status,detail,checked_at FROM sense_admission_results WHERE device_id=$1`, deviceID).Scan(&result.DeviceID, &result.Address, &result.Status, &result.Detail, &result.CheckedAt)
err := s.database.QueryRowContext(ctx, `SELECT device_id,address,status,detail,media_status,media_detail,checked_at FROM sense_admission_results WHERE device_id=$1`, deviceID).Scan(&result.DeviceID, &result.Address, &result.Status, &result.Detail, &result.MediaStatus, &result.MediaDetail, &result.CheckedAt)
if errors.Is(err, sql.ErrNoRows) {
return Result{}, ErrNotFound
}
@@ -11,6 +11,11 @@ type Credential struct {
Password string
}
type DisplayInfo struct {
Name string
Location string
}
var activeService atomic.Pointer[Service]
// ReadCredential is an internal adapter port. Credentials must never be
@@ -50,6 +55,18 @@ func ReadRTSPCredential(ctx context.Context, id string) (Credential, error) {
return Credential{Username: username, Password: password}, err
}
func Describe(ctx context.Context, id string) (DisplayInfo, error) {
service := activeService.Load()
if service == nil {
return DisplayInfo{}, fmt.Errorf("device service is not ready")
}
item, err := service.store.Get(ctx, id)
if err != nil {
return DisplayInfo{}, err
}
return DisplayInfo{Name: item.Name, Location: item.Location}, nil
}
func MarkActive(ctx context.Context, id string) error {
service := activeService.Load()
if service == nil {
+39 -4
View File
@@ -10,15 +10,34 @@ import (
"sync"
"time"
"yovision.local/sense/app/sense/admission"
"yovision.local/sense/app/sense/device"
"yovision.local/sense/app/sense/media"
)
type Route struct {
ID string `json:"id"`
DeviceID string `json:"device_id"`
DeviceName string `json:"device_name"`
DeviceLocation string `json:"device_location"`
ProfileToken string `json:"profile_token"`
ProfileName string `json:"profile_name"`
ProfileKind string `json:"profile_kind"`
Desired string `json:"desired"`
Actual string `json:"actual"`
Detail string `json:"detail"`
Readers int `json:"readers"`
}
type Session struct {
ID string `json:"id"`
OwnerID string `json:"-"`
RouteID string `json:"route_id"`
DeviceID string `json:"device_id"`
DeviceName string `json:"device_name"`
ProfileToken string `json:"profile_token"`
ProfileName string `json:"profile_name"`
ProfileKind string `json:"profile_kind"`
PlayerURL string `json:"player_url"`
Status string `json:"status"`
Detail string `json:"detail"`
@@ -47,15 +66,24 @@ func NewService(rawBase string, ttl time.Duration) (*Service, error) {
}
return &Service{base: parsed, ttl: ttl, route: media.PlaybackRoute, routes: media.PlaybackRoutes, sessions: map[string]Session{}, now: time.Now}, nil
}
func (s *Service) Routes(ctx context.Context) ([]media.Route, error) {
func (s *Service) Routes(ctx context.Context) ([]Route, error) {
items, err := s.routes(ctx)
if err != nil {
return nil, err
}
result := items[:0]
result := make([]Route, 0, len(items))
for _, item := range items {
if item.Desired == "running" {
result = append(result, item)
view := Route{ID: item.ID, DeviceID: item.DeviceID, DeviceName: item.DeviceID, ProfileToken: item.ProfileToken, ProfileName: item.ProfileToken, Desired: item.Desired, Actual: item.Actual, Detail: item.Detail, Readers: item.Readers}
if info, describeErr := device.Describe(ctx, item.DeviceID); describeErr == nil {
view.DeviceName = info.Name
view.DeviceLocation = info.Location
}
if profile, profileErr := admission.VerifiedProfile(item.DeviceID, item.ProfileToken); profileErr == nil {
view.ProfileName = profile.Name
view.ProfileKind = profile.Kind
}
result = append(result, view)
}
}
return result, nil
@@ -69,7 +97,14 @@ func (s *Service) Create(ctx context.Context, owner, routeID string) (Session, e
return Session{}, fmt.Errorf("media route is stopped")
}
id := newID()
session := Session{ID: id, OwnerID: owner, RouteID: route.ID, DeviceID: route.DeviceID, ProfileToken: route.ProfileToken, PlayerURL: "/api/v1/liveview/sessions/" + id + "/player", Status: route.Actual, Detail: route.Detail, ExpiresAt: s.now().UTC().Add(s.ttl)}
session := Session{ID: id, OwnerID: owner, RouteID: route.ID, DeviceID: route.DeviceID, DeviceName: route.DeviceID, ProfileToken: route.ProfileToken, ProfileName: route.ProfileToken, PlayerURL: "/api/v1/liveview/sessions/" + id + "/player", Status: route.Actual, Detail: route.Detail, ExpiresAt: s.now().UTC().Add(s.ttl)}
if info, describeErr := device.Describe(ctx, route.DeviceID); describeErr == nil {
session.DeviceName = info.Name
}
if profile, profileErr := admission.VerifiedProfile(route.DeviceID, route.ProfileToken); profileErr == nil {
session.ProfileName = profile.Name
session.ProfileKind = profile.Kind
}
s.mu.Lock()
s.sessions[id] = session
s.mu.Unlock()
@@ -4,6 +4,9 @@ import (
"context"
"fmt"
"sync/atomic"
"yovision.local/sense/app/sense/admission"
"yovision.local/sense/app/sense/identity"
)
var activeService atomic.Pointer[Service]
@@ -23,3 +26,11 @@ func PlaybackRoutes(ctx context.Context) ([]Route, error) {
}
return service.store.List(ctx)
}
func ConfigureReadyProfiles(ctx context.Context, actor identity.Principal, result admission.Result) admission.MediaOutcome {
service := activeService.Load()
if service == nil {
return admission.MediaOutcome{Status: "needs_attention", Detail: "摄像机已接入,媒体服务尚未就绪"}
}
return service.ConfigureProfiles(ctx, actor, result)
}
+29 -3
View File
@@ -21,7 +21,33 @@ type Service struct {
}
func NewService(store Store, process mediamtx.Process, controller mediamtx.Controller) *Service {
return &Service{store: store, process: process, controller: controller, profile: admission.VerifiedProfile, credential: device.ReadCredential, now: time.Now}
return &Service{store: store, process: process, controller: controller, profile: admission.VerifiedProfile, credential: device.ReadRTSPCredential, now: time.Now}
}
func (s *Service) ConfigureProfiles(ctx context.Context, actor identity.Principal, result admission.Result) admission.MediaOutcome {
configured := 0
ready := 0
for _, profile := range result.Profiles {
if profile.Verification.Status != "ready" {
continue
}
configured++
route, err := s.Configure(ctx, actor, result.DeviceID, profile.Token)
if err != nil {
continue
}
route, err = s.Reconcile(ctx, actor, route.ID)
if err == nil && route.Actual == "ready" {
ready++
}
}
if configured == 0 {
return admission.MediaOutcome{Status: "needs_attention", Detail: "未找到可用码流,请检查接入结果"}
}
if ready == configured {
return admission.MediaOutcome{Status: "ready", Detail: "视频已进入实时监看"}
}
return admission.MediaOutcome{Status: "needs_attention", Detail: "摄像机已接入,媒体服务未就绪,请到视频服务查看并重试"}
}
func (s *Service) Configure(ctx context.Context, actor identity.Principal, deviceID, profileToken string) (Route, error) {
if _, err := s.profile(deviceID, profileToken); err != nil {
@@ -55,7 +81,7 @@ func (s *Service) Reconcile(ctx context.Context, actor identity.Principal, id st
}
if err := s.process.Start(ctx); err != nil {
route.Actual = "process_failed"
route.Detail = err.Error()
route.Detail = "媒体进程未启动,请检查视频服务配置"
route.Version++
route.UpdatedAt = s.now().UTC()
_ = s.store.Save(ctx, route)
@@ -72,7 +98,7 @@ func (s *Service) Reconcile(ctx context.Context, actor identity.Principal, id st
source := mediamtx.Source{Path: route.Path, URI: profile.StreamURI, Username: credential.Username, Password: credential.Password}
if err := s.controller.Apply(ctx, source); err != nil {
route.Actual = "apply_failed"
route.Detail = err.Error()
route.Detail = "媒体路径配置失败,请检查视频服务"
route.Version++
route.UpdatedAt = s.now().UTC()
_ = s.store.Save(ctx, route)
@@ -3,9 +3,11 @@ package media
import (
"context"
"errors"
"strings"
"testing"
"yovision.local/sense/app/sense/adapters/mediamtx"
"yovision.local/sense/app/sense/adapters/rtsp"
"yovision.local/sense/app/sense/admission"
"yovision.local/sense/app/sense/device"
"yovision.local/sense/app/sense/identity"
@@ -108,3 +110,36 @@ func TestReconcileRecordsFailureStates(t *testing.T) {
t.Fatalf("result=%#v err=%v", result, err)
}
}
func TestConfigureProfilesIsIdempotentAndReportsMediaState(t *testing.T) {
process := &fakeProcess{startErr: errors.New("binary missing")}
service := NewService(NewMemoryStore(), process, fakeController{})
service.profile = func(_ string, token string) (admission.Profile, error) {
return admission.Profile{Token: token, StreamURI: "rtsp://camera.invalid/" + token}, nil
}
service.credential = func(context.Context, string) (device.Credential, error) {
return device.Credential{Username: "fixture", Password: "fixture"}, nil
}
result := admission.Result{DeviceID: "device", Profiles: []admission.Profile{{Token: "main", Verification: rtspReady()}, {Token: "sub", Verification: rtspReady()}}}
outcome := service.ConfigureProfiles(context.Background(), identity.Principal{}, result)
if outcome.Status != "needs_attention" {
t.Fatalf("outcome=%#v", outcome)
}
outcome = service.ConfigureProfiles(context.Background(), identity.Principal{}, result)
items, err := service.List(context.Background())
if err != nil || len(items) != 2 || outcome.Status != "needs_attention" {
t.Fatalf("items=%#v outcome=%#v err=%v", items, outcome, err)
}
for _, item := range items {
if item.Actual != "process_failed" || strings.Contains(item.Detail, "binary missing") {
t.Fatalf("unsafe route detail=%#v", item)
}
}
process.startErr = nil
outcome = service.ConfigureProfiles(context.Background(), identity.Principal{}, result)
if outcome.Status != "ready" {
t.Fatalf("ready outcome=%#v", outcome)
}
}
func rtspReady() rtsp.Result { return rtsp.Result{Status: "ready"} }
@@ -6,6 +6,7 @@ import (
"yovision.local/sense/app/sense/adapters/onvif"
"yovision.local/sense/app/sense/adapters/rtsp"
"yovision.local/sense/app/sense/admission"
"yovision.local/sense/app/sense/media"
"yovision.local/sense/internal/platform"
)
@@ -18,7 +19,9 @@ func init() {
store = admission.NewPostgresStore(app.Database())
}
app.RegisterMigration(platform.Migration{Version: 2026081302, Name: "sense_admission_profiles", SQL: admission.MigrationSQL})
app.RegisterMigration(platform.Migration{Version: 2026081303, Name: "sense_admission_media_status", SQL: admission.MediaStatusMigrationSQL})
service := admission.NewService(onvif.NewHTTPClient(8*time.Second), rtsp.NetVerifier{Timeout: 5 * time.Second}, os.Getenv("SENSE_ONVIF_DISCOVERY_IP"), store)
service.SetReadyHandler(media.ConfigureReadyProfiles)
admission.NewModule(service).Register(app)
return nil
})
@@ -19,6 +19,7 @@
<el-empty v-if="!result.status" description="选择设备并开始检查" />
<template v-else>
<el-result :icon="result.status === 'ready' ? 'success' : 'warning'" :title="statusLabel(result.status)" :sub-title="result.detail" />
<el-alert v-if="result.media_status" :title="result.media_detail" :type="result.media_status === 'ready' ? 'success' : 'warning'" show-icon :closable="false" class="media-result" />
<el-table v-if="result.profiles?.length" :data="result.profiles" border>
<el-table-column label="用途" width="90"><template #default="scope"><el-tag>{{ scope.row.kind === 'main' ? '主码流' : scope.row.kind === 'sub' ? '子码流' : '其他' }}</el-tag></template></el-table-column>
<el-table-column prop="name" label="Profile" min-width="120" />
@@ -50,4 +51,4 @@ async function discover(){discovering.value=true;try{discovered.value=(await dis
function useAddress(value){form.address=value;discoveryDialog.value=false}
onMounted(loadDevices)
</script>
<style scoped>.admission-form{margin-top:18px}.field-hint{color:#86909c;font-size:12px;line-height:1.5}.result-header{display:flex;align-items:center;justify-content:space-between}</style>
<style scoped>.admission-form{margin-top:18px}.field-hint{color:#86909c;font-size:12px;line-height:1.5}.result-header{display:flex;align-items:center;justify-content:space-between}.media-result{margin-bottom:16px}</style>
@@ -2,14 +2,14 @@
<section class="page-container">
<div class="page-heading"><div><h1>实时监看</h1><p>一次打开一路视频,可在已验证的主、子码流之间切换。</p></div><el-tag v-if="session.expires_at" type="info">会话短期有效</el-tag></div>
<el-row :gutter="16">
<el-col :span="7"><el-card shadow="never"><template #header><strong>选择视频</strong></template><el-form label-position="top"><el-form-item label="设备与 Profile"><el-select v-model="selectedRoute" filterable placeholder="选择一路视频" style="width:100%" @change="open"><el-option v-for="item in routes" :key="item.id" :label="`${item.device_id} · ${profileLabel(item.profile_token)}`" :value="item.id"><span>{{ item.device_id }}</span><span class="option-detail">{{ profileLabel(item.profile_token) }} · {{ stateLabel(item.actual) }}</span></el-option></el-select></el-form-item></el-form><el-descriptions v-if="current" :column="1" border><el-descriptions-item label="设备">{{ current.device_id }}</el-descriptions-item><el-descriptions-item label="Profile">{{ profileLabel(current.profile_token) }}</el-descriptions-item><el-descriptions-item label="媒体状态"><el-tag :type="current.actual === 'ready' ? 'success' : 'warning'">{{ stateLabel(current.actual) }}</el-tag></el-descriptions-item><el-descriptions-item label="观看连接">{{ current.readers }}</el-descriptions-item></el-descriptions><el-alert v-else title="先在视频服务中建立并对账媒体路径" type="info" show-icon :closable="false" /></el-card></el-col>
<el-col :span="7"><el-card shadow="never"><template #header><strong>选择视频</strong></template><el-form label-position="top"><el-form-item label="设备与码流"><el-select v-model="selectedRoute" filterable placeholder="选择一路视频" style="width:100%" @change="open"><el-option v-for="item in routes" :key="item.id" :label="`${item.device_name} · ${profileLabel(item)}`" :value="item.id"><span>{{ item.device_name }}</span><span class="option-detail">{{ profileLabel(item) }} · {{ stateLabel(item.actual) }}</span></el-option></el-select></el-form-item></el-form><el-descriptions v-if="current" :column="1" border><el-descriptions-item label="设备">{{ current.device_name }}</el-descriptions-item><el-descriptions-item label="位置">{{ current.device_location || '未填写' }}</el-descriptions-item><el-descriptions-item label="码流">{{ profileLabel(current) }}</el-descriptions-item><el-descriptions-item label="媒体状态"><el-tag :type="current.actual === 'ready' ? 'success' : 'warning'">{{ stateLabel(current.actual) }}</el-tag></el-descriptions-item><el-descriptions-item label="观看连接">{{ current.readers }}</el-descriptions-item></el-descriptions><el-alert v-else title="请先在视频接入完成检查;系统会自动建立媒体路径。若仍无视频,请到视频服务排错。" type="info" show-icon :closable="false" /></el-card></el-col>
<el-col :span="17"><el-card shadow="never"><template #header><div class="player-header"><strong>单路画面</strong><el-button v-if="selectedRoute" :icon="Refresh" @click="retry">重新连接</el-button></div></template><StreamPlayer :state="playerState" :detail="playerDetail" :player-url="session.player_url" @retry="retry" /></el-card></el-col>
</el-row>
</section>
</template>
<script setup>
import{computed,onBeforeUnmount,onMounted,reactive,ref}from'vue';import{ElMessage}from'element-plus';import{Refresh}from'@element-plus/icons-vue';import{createSession,getSession,playbackRoutes}from'../../../api/sense/liveview';import StreamPlayer from'../../../components/sense/liveview/StreamPlayer.vue'
const routes=ref([]),selectedRoute=ref(''),session=reactive({}),playerState=ref('waiting'),playerDetail=ref('请选择一路视频'),timer=ref();const current=computed(()=>routes.value.find(item=>item.id===selectedRoute.value));function profileLabel(value){return value?.toLowerCase().includes('sub')?'子码流':value?.toLowerCase().includes('main')?'主码流':value}function stateLabel(value){return({ready:'正常',waiting:'等待视频',stopped:'已停止',process_failed:'服务未启动',apply_failed:'配置失败',unconverged:'状态未同步'})[value]||value}
const routes=ref([]),selectedRoute=ref(''),session=reactive({}),playerState=ref('waiting'),playerDetail=ref('请选择一路视频'),timer=ref();const current=computed(()=>routes.value.find(item=>item.id===selectedRoute.value));function profileLabel(item){const kind=({main:'主码流',sub:'子码流',other:'其他码流'})[item?.profile_kind];return kind?`${kind}${item.profile_name?`(${item.profile_name})`:''}`:(item?.profile_name||item?.profile_token)}function stateLabel(value){return({ready:'正常',waiting:'等待视频',stopped:'已停止',process_failed:'服务未启动',apply_failed:'配置失败',unconverged:'状态未同步'})[value]||value}
async function load(){routes.value=(await playbackRoutes()).items||[]}
async function open(){if(!selectedRoute.value)return;playerState.value='loading';try{Object.assign(session,await createSession(selectedRoute.value));playerState.value=session.status==='ready'?'ready':session.status;playerDetail.value=session.detail;startPolling()}catch(error){playerState.value='offline';playerDetail.value=error.message||'无法创建播放会话'}}
function startPolling(){clearInterval(timer.value);timer.value=setInterval(async()=>{if(!session.id)return;try{const latest=await getSession(session.id);Object.assign(session,latest);playerState.value=latest.status==='ready'?'ready':latest.status;playerDetail.value=latest.detail}catch{playerState.value='expired';playerDetail.value='播放会话已过期,请重新连接';clearInterval(timer.value)}},5000)}