diff --git a/Sense/server/app/sense/admission/migration.go b/Sense/server/app/sense/admission/migration.go index fab2b30..1eb8b60 100644 --- a/Sense/server/app/sense/admission/migration.go +++ b/Sense/server/app/sense/admission/migration.go @@ -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 ''; +` diff --git a/Sense/server/app/sense/admission/service.go b/Sense/server/app/sense/admission/service.go index d1abbeb..2edeb43 100644 --- a/Sense/server/app/sense/admission/service.go +++ b/Sense/server/app/sense/admission/service.go @@ -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 } diff --git a/Sense/server/app/sense/admission/service_test.go b/Sense/server/app/sense/admission/service_test.go index 411d833..35f2249 100644 --- a/Sense/server/app/sense/admission/service_test.go +++ b/Sense/server/app/sense/admission/service_test.go @@ -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) { diff --git a/Sense/server/app/sense/admission/store.go b/Sense/server/app/sense/admission/store.go index 47a3ff3..b1e6f65 100644 --- a/Sense/server/app/sense/admission/store.go +++ b/Sense/server/app/sense/admission/store.go @@ -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 } diff --git a/Sense/server/app/sense/device/credential_port.go b/Sense/server/app/sense/device/credential_port.go index e0071ec..58f8014 100644 --- a/Sense/server/app/sense/device/credential_port.go +++ b/Sense/server/app/sense/device/credential_port.go @@ -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 { diff --git a/Sense/server/app/sense/liveview/service.go b/Sense/server/app/sense/liveview/service.go index 101ee80..454ecd7 100644 --- a/Sense/server/app/sense/liveview/service.go +++ b/Sense/server/app/sense/liveview/service.go @@ -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() diff --git a/Sense/server/app/sense/media/route_port.go b/Sense/server/app/sense/media/route_port.go index 59596f8..024f88f 100644 --- a/Sense/server/app/sense/media/route_port.go +++ b/Sense/server/app/sense/media/route_port.go @@ -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) +} diff --git a/Sense/server/app/sense/media/service.go b/Sense/server/app/sense/media/service.go index 8197930..666009c 100644 --- a/Sense/server/app/sense/media/service.go +++ b/Sense/server/app/sense/media/service.go @@ -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) diff --git a/Sense/server/app/sense/media/service_test.go b/Sense/server/app/sense/media/service_test.go index 5f536d3..612175d 100644 --- a/Sense/server/app/sense/media/service_test.go +++ b/Sense/server/app/sense/media/service_test.go @@ -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"} } diff --git a/Sense/server/cmd/sense/modules_admission.go b/Sense/server/cmd/sense/modules_admission.go index 7771243..d2c260a 100644 --- a/Sense/server/cmd/sense/modules_admission.go +++ b/Sense/server/cmd/sense/modules_admission.go @@ -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 }) diff --git a/Sense/ui/src/views/sense/admission/Admission.vue b/Sense/ui/src/views/sense/admission/Admission.vue index c89b557..f58448f 100644 --- a/Sense/ui/src/views/sense/admission/Admission.vue +++ b/Sense/ui/src/views/sense/admission/Admission.vue @@ -19,6 +19,7 @@