yovision/Sense/server/app/sense/admission/service.go

package admission

import (
	"context"
	"errors"
	"fmt"
	"os"
	"sort"
	"strings"
	"time"

	coreService "github.com/go-admin-team/go-admin-core/sdk/service"
	"gorm.io/gorm"

	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/area"
	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/credential"
	deviceModels "git.ilapage.cn/ila/yovision/Sense/server/app/sense/device/models"
	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/onvif"
	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/rtsp"
)

var (
	ErrNotFound = errors.New("尚无该设备的接入结果")
	ErrInvalid  = errors.New("接入请求不符合要求")
	ErrConflict = errors.New("设备已被其他用户更新，请刷新后重试")
)

type Service struct {
	coreService.Service
	ONVIF       onvif.Client
	RTSP        rtsp.Verifier
	Policy      onvif.Policy
	DiscoveryIP string
	Timeout     time.Duration
}

func NewRuntime(service coreService.Service) (*Service, error) {
	policy, err := onvif.ParsePolicy(os.Getenv("SENSE_ONVIF_ALLOWED_CIDRS"))
	if err != nil {
		return nil, err
	}
	return &Service{Service: service, Policy: policy, DiscoveryIP: strings.TrimSpace(os.Getenv("SENSE_ONVIF_DISCOVERY_IP")), ONVIF: onvif.NewHTTPClient(8*time.Second, policy), RTSP: rtsp.NetVerifier{Timeout: 5 * time.Second, Policy: policy}}, nil
}
func (s *Service) Discover(ctx context.Context) (DiscoveryResponse, error) {
	addresses, err := onvif.Discover(ctx, s.DiscoveryIP, 3*time.Second, s.Policy)
	return DiscoveryResponse{Addresses: addresses, Interface: s.DiscoveryIP}, err
}
func (s *Service) Probe(ctx context.Context, request ProbeRequest) (ResultResponse, error) {
	if request.Version < 1 || strings.TrimSpace(request.Address) == "" || len(request.Address) > 1024 {
		return ResultResponse{}, ErrInvalid
	}
	var device deviceModels.Device
	if err := s.Orm.Select("id", "modality", "version").First(&device, "id = ?", request.DeviceID).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return ResultResponse{}, ErrNotFound
		}
		return ResultResponse{}, err
	}
	if device.Modality != deviceModels.ModalityVideo {
		return ResultResponse{}, ErrInvalid
	}
	if device.Version != request.Version {
		return ResultResponse{}, ErrConflict
	}
	onvifValue, err := credential.Read(s.Orm, request.DeviceID, credential.PurposeONVIF)
	if err != nil {
		return ResultResponse{}, err
	}
	rtspValue, err := credential.Read(s.Orm, request.DeviceID, credential.PurposeRTSP)
	if err != nil {
		return ResultResponse{}, err
	}
	profiles, probeErr := s.ONVIF.Profiles(ctx, request.Address, onvif.Credential{Username: onvifValue.Username, Password: onvifValue.Password})
	now := time.Now().UTC()
	result := Result{DeviceID: request.DeviceID, Address: strings.TrimSpace(request.Address), CheckedAt: now, UpdatedAt: now}
	if probeErr != nil {
		result.Status, result.Detail = classify(probeErr)
		if err = s.save(result, request, false); err != nil {
			return ResultResponse{}, err
		}
		return s.Get(request.DeviceID)
	}
	for _, profile := range profiles {
		verification, verifyErr := s.RTSP.Verify(ctx, profile.StreamURI, rtsp.Credential{Username: rtspValue.Username, Password: rtspValue.Password})
		if verifyErr != nil {
			verification = rtsp.Result{Status: "failed", Detail: "视频地址未通过安全检查"}
		}
		result.Profiles = append(result.Profiles, Profile{DeviceID: request.DeviceID, Token: profile.Token, Name: profile.Name, Width: profile.Width, Height: profile.Height, Encoding: profile.Encoding, StreamURI: profile.StreamURI, Kind: "other", VerificationStatus: verification.Status, VerificationLatencyMS: verification.LatencyMS, VerificationDetail: verification.Detail, UpdatedAt: now})
	}
	sort.Slice(result.Profiles, func(i, j int) bool {
		return result.Profiles[i].Width*result.Profiles[i].Height > result.Profiles[j].Width*result.Profiles[j].Height
	})
	if len(result.Profiles) > 0 {
		result.Profiles[0].Kind = "main"
	}
	if len(result.Profiles) > 1 {
		result.Profiles[len(result.Profiles)-1].Kind = "sub"
	}
	result.Status = "ready"
	result.Detail = "设备与视频 Profile 已验证"
	for _, profile := range result.Profiles {
		if profile.VerificationStatus != "ready" {
			result.Status = "profile_failed"
			result.Detail = "部分视频 Profile 验证失败"
		}
	}
	if err = s.save(result, request, true); err != nil {
		return ResultResponse{}, err
	}
	return response(result), nil
}
func (s *Service) save(result Result, request ProbeRequest, replaceProfiles bool) error {
	return s.Orm.Transaction(func(tx *gorm.DB) error {
		updates := map[string]any{"version": request.Version + 1, "update_by": request.UpdateBy, "updated_at": result.UpdatedAt, "retry_requested_at": nil}
		if replaceProfiles {
			updates["status"] = map[bool]string{true: deviceModels.StatusActive, false: deviceModels.StatusPending}[result.Status == "ready"]
			updates["adapter_status"] = map[bool]string{true: deviceModels.AdapterReady, false: deviceModels.AdapterFailed}[result.Status == "ready"]
		}
		update := tx.Model(&deviceModels.Device{}).Where("id = ? AND version = ?", request.DeviceID, request.Version).Updates(updates)
		if update.Error != nil {
			return update.Error
		}
		if update.RowsAffected == 0 {
			return ErrConflict
		}
		if err := tx.Omit("Profiles").Save(&result).Error; err != nil {
			return err
		}
		if replaceProfiles {
			if err := tx.Where("device_id = ?", request.DeviceID).Delete(&Profile{}).Error; err != nil {
				return err
			}
		}
		if replaceProfiles && len(result.Profiles) > 0 {
			if err := tx.Create(&result.Profiles).Error; err != nil {
				return err
			}
		}
		if replaceProfiles {
			snapshots := make([]area.ProfileSnapshot, 0, len(result.Profiles))
			for _, profile := range result.Profiles {
				snapshots = append(snapshots, area.ProfileSnapshot{Token: profile.Token, Width: profile.Width, Height: profile.Height, Encoding: profile.Encoding})
			}
			return area.MarkProfilesReplaced(tx, request.DeviceID, snapshots)
		}
		return nil
	})
}
func (s *Service) Get(deviceID string) (ResultResponse, error) {
	var result Result
	if err := s.Orm.Preload("Profiles", func(db *gorm.DB) *gorm.DB { return db.Order("width * height DESC") }).First(&result, "device_id = ?", deviceID).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return ResultResponse{}, ErrNotFound
		}
		return ResultResponse{}, fmt.Errorf("read admission: %w", err)
	}
	return response(result), nil
}
func response(result Result) ResultResponse {
	out := ResultResponse{DeviceID: result.DeviceID, Address: result.Address, Status: result.Status, Detail: result.Detail, CheckedAt: result.CheckedAt, Profiles: make([]ProfileResponse, 0, len(result.Profiles))}
	for _, p := range result.Profiles {
		out.Profiles = append(out.Profiles, ProfileResponse{Token: p.Token, Name: p.Name, Width: p.Width, Height: p.Height, Encoding: p.Encoding, StreamURI: p.StreamURI, Kind: p.Kind, VerificationStatus: p.VerificationStatus, VerificationLatencyMS: p.VerificationLatencyMS, VerificationDetail: p.VerificationDetail})
	}
	return out
}
func classify(err error) (string, string) {
	switch {
	case errors.Is(err, onvif.ErrAuthentication):
		return "authentication_failed", "设备拒绝了当前凭据，请更新后重试"
	case errors.Is(err, onvif.ErrTargetNotAllowed):
		return "target_not_allowed", "设备地址不在获准网段内"
	case errors.Is(err, onvif.ErrRedirect):
		return "redirect_rejected", "设备返回了不允许的重定向"
	case errors.Is(err, context.DeadlineExceeded) || strings.Contains(strings.ToLower(err.Error()), "timeout"):
		return "timeout", "设备响应超时"
	case strings.Contains(strings.ToLower(err.Error()), "time") || strings.Contains(strings.ToLower(err.Error()), "clock"):
		return "clock_skew", "设备时间可能不准确，请校时后重试"
	default:
		return "unreachable", "无法读取设备信息，请检查地址和网络"
	}
}