673 lines
23 KiB
Go
673 lines
23 KiB
Go
package platform
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"sort"
|
||
"strings"
|
||
"time"
|
||
)
|
||
|
||
type accessEvidenceStore interface {
|
||
AccessEvidence(context.Context) ([]AccessEvidenceRow, error)
|
||
}
|
||
|
||
type accessThresholdStore interface {
|
||
AccessThresholds(context.Context) (AccessThresholdConfig, error)
|
||
SaveAccessThresholds(context.Context, AccessThresholdUpdate) (AccessThresholdConfig, error)
|
||
}
|
||
|
||
type accessUnresolvedIdentityStore interface {
|
||
AccessUnresolvedIdentities(context.Context, AccessUnresolvedIdentityQuery) (Page[AccessUnresolvedIdentity], error)
|
||
}
|
||
|
||
func defaultAccessThresholds(now time.Time) AccessThresholdConfig {
|
||
return AccessThresholdConfig{
|
||
Version: 1,
|
||
DefaultThresholdSec: 300,
|
||
DelayThresholdSec: 30,
|
||
LongOfflineSec: 1800,
|
||
Protocols: []AccessProtocolThreshold{
|
||
{Protocol: "GB32960", ThresholdSec: 300},
|
||
{Protocol: "JT808", ThresholdSec: 300},
|
||
{Protocol: "YUTONG_MQTT", ThresholdSec: 600},
|
||
},
|
||
UpdatedBy: "system",
|
||
UpdatedAt: now.Format(time.RFC3339),
|
||
}
|
||
}
|
||
|
||
func (s *Service) AccessThresholds(ctx context.Context) (AccessThresholdConfig, error) {
|
||
store, ok := s.store.(accessThresholdStore)
|
||
if !ok {
|
||
return defaultAccessThresholds(time.Now()), nil
|
||
}
|
||
config, err := store.AccessThresholds(ctx)
|
||
if err != nil {
|
||
return AccessThresholdConfig{}, err
|
||
}
|
||
return normalizeAccessThresholdConfig(config, time.Now()), nil
|
||
}
|
||
|
||
func (s *Service) UpdateAccessThresholds(ctx context.Context, update AccessThresholdUpdate) (AccessThresholdConfig, error) {
|
||
if err := validateAccessThresholdUpdate(update); err != nil {
|
||
return AccessThresholdConfig{}, err
|
||
}
|
||
store, ok := s.store.(accessThresholdStore)
|
||
if !ok {
|
||
return AccessThresholdConfig{}, clientError{Code: "ACCESS_THRESHOLD_READ_ONLY", Message: "当前存储不支持更新接入阈值"}
|
||
}
|
||
update.Actor = strings.TrimSpace(update.Actor)
|
||
if update.Actor == "" {
|
||
update.Actor = "platform-admin"
|
||
}
|
||
return store.SaveAccessThresholds(ctx, update)
|
||
}
|
||
|
||
func (s *Service) AccessVehicles(ctx context.Context, query AccessQuery) (Page[AccessVehicleRow], error) {
|
||
rows, config, err := s.accessRows(ctx, query)
|
||
if err != nil {
|
||
return Page[AccessVehicleRow]{}, err
|
||
}
|
||
_ = config
|
||
limit := query.Limit
|
||
if limit <= 0 {
|
||
limit = 50
|
||
}
|
||
if limit > 200 {
|
||
limit = 200
|
||
}
|
||
offset := query.Offset
|
||
if offset < 0 {
|
||
offset = 0
|
||
}
|
||
total := len(rows)
|
||
if offset >= total {
|
||
return Page[AccessVehicleRow]{Items: []AccessVehicleRow{}, Total: total, Limit: limit, Offset: offset}, nil
|
||
}
|
||
end := offset + limit
|
||
if end > total {
|
||
end = total
|
||
}
|
||
return Page[AccessVehicleRow]{Items: append([]AccessVehicleRow(nil), rows[offset:end]...), Total: total, Limit: limit, Offset: offset}, nil
|
||
}
|
||
|
||
func (s *Service) AccessUnresolvedIdentities(ctx context.Context, query AccessUnresolvedIdentityQuery) (Page[AccessUnresolvedIdentity], error) {
|
||
store, ok := s.store.(accessUnresolvedIdentityStore)
|
||
if !ok {
|
||
return Page[AccessUnresolvedIdentity]{}, fmt.Errorf("store does not provide unresolved identity evidence")
|
||
}
|
||
query.Keyword = strings.TrimSpace(query.Keyword)
|
||
query.Protocol = strings.ToUpper(strings.TrimSpace(query.Protocol))
|
||
if query.Protocol != "" && query.Protocol != "JT808" {
|
||
return Page[AccessUnresolvedIdentity]{Items: []AccessUnresolvedIdentity{}, Limit: 50}, nil
|
||
}
|
||
if query.Limit <= 0 {
|
||
query.Limit = 50
|
||
}
|
||
if query.Limit > 200 {
|
||
query.Limit = 200
|
||
}
|
||
if query.Offset < 0 {
|
||
query.Offset = 0
|
||
}
|
||
return store.AccessUnresolvedIdentities(ctx, query)
|
||
}
|
||
|
||
func (s *Service) AccessSummary(ctx context.Context, query AccessQuery) (AccessSummary, error) {
|
||
rows, config, err := s.accessRows(ctx, query)
|
||
if err != nil {
|
||
return AccessSummary{}, err
|
||
}
|
||
now := time.Now()
|
||
result := AccessSummary{TotalVehicles: len(rows), AsOf: now.Format(time.RFC3339), ThresholdVersion: config.Version}
|
||
protocols := map[string]*AccessDistribution{}
|
||
oems := map[string]*AccessDistribution{}
|
||
for _, row := range rows {
|
||
switch row.ConnectionState {
|
||
case "healthy":
|
||
result.HealthyVehicles++
|
||
case "incomplete":
|
||
result.IncompleteVehicles++
|
||
case "degraded":
|
||
result.DegradedVehicles++
|
||
}
|
||
switch row.OnlineState {
|
||
case "online":
|
||
result.OnlineVehicles++
|
||
case "offline":
|
||
result.OfflineVehicles++
|
||
if row.FreshnessSec != nil && *row.FreshnessSec >= config.LongOfflineSec {
|
||
result.LongOfflineVehicles++
|
||
}
|
||
case "never_reported":
|
||
result.NeverReported++
|
||
default:
|
||
result.UnknownVehicles++
|
||
}
|
||
if row.DelayAbnormal {
|
||
result.DelayAbnormal++
|
||
}
|
||
if reportedOnDay(row.LatestReceivedAt, now) {
|
||
result.ReportedToday++
|
||
}
|
||
for _, status := range row.ProtocolStatuses {
|
||
if status.Connected {
|
||
addAccessDistribution(protocols, status.Protocol, status.OnlineState == "online")
|
||
}
|
||
}
|
||
addAccessDistribution(oems, firstNonEmpty(row.OEM, "未维护"), row.OnlineState == "online")
|
||
}
|
||
if result.TotalVehicles > 0 {
|
||
result.OnlineRate = float64(result.OnlineVehicles) / float64(result.TotalVehicles) * 100
|
||
}
|
||
result.Protocols = sortedAccessDistributions(protocols)
|
||
result.OEMs = sortedAccessDistributions(oems)
|
||
return result, nil
|
||
}
|
||
|
||
func (s *Service) accessRows(ctx context.Context, query AccessQuery) ([]AccessVehicleRow, AccessThresholdConfig, error) {
|
||
if err := validateAccessQuery(query); err != nil {
|
||
return nil, AccessThresholdConfig{}, err
|
||
}
|
||
store, ok := s.store.(accessEvidenceStore)
|
||
if !ok {
|
||
return nil, AccessThresholdConfig{}, fmt.Errorf("store does not provide access evidence")
|
||
}
|
||
config, err := s.AccessThresholds(ctx)
|
||
if err != nil {
|
||
return nil, AccessThresholdConfig{}, err
|
||
}
|
||
evidence, err := store.AccessEvidence(ctx)
|
||
if err != nil {
|
||
return nil, AccessThresholdConfig{}, err
|
||
}
|
||
now := time.Now()
|
||
byVIN := make(map[string][]AccessEvidenceRow, len(evidence))
|
||
order := make([]string, 0, len(evidence))
|
||
for _, item := range evidence {
|
||
vin := strings.TrimSpace(item.VIN)
|
||
if _, exists := byVIN[vin]; !exists {
|
||
order = append(order, vin)
|
||
}
|
||
byVIN[vin] = append(byVIN[vin], item)
|
||
}
|
||
rows := make([]AccessVehicleRow, 0, len(byVIN))
|
||
for _, vin := range order {
|
||
row := buildAccessVehicleGroup(byVIN[vin], config, now)
|
||
if keepAccessRow(row, query) {
|
||
rows = append(rows, row)
|
||
}
|
||
}
|
||
sort.SliceStable(rows, func(i, j int) bool {
|
||
left, right := accessStateRank(rows[i].OnlineState), accessStateRank(rows[j].OnlineState)
|
||
if left != right {
|
||
return left < right
|
||
}
|
||
if rows[i].LatestReceivedAt != rows[j].LatestReceivedAt {
|
||
return rows[i].LatestReceivedAt > rows[j].LatestReceivedAt
|
||
}
|
||
return rows[i].VIN < rows[j].VIN
|
||
})
|
||
return rows, config, nil
|
||
}
|
||
|
||
func buildAccessVehicleGroup(items []AccessEvidenceRow, config AccessThresholdConfig, now time.Time) AccessVehicleRow {
|
||
if len(items) == 0 {
|
||
return AccessVehicleRow{}
|
||
}
|
||
base := items[0]
|
||
row := AccessVehicleRow{
|
||
VIN: strings.TrimSpace(base.VIN),
|
||
Plate: strings.TrimSpace(base.Plate),
|
||
OEM: strings.TrimSpace(base.OEM),
|
||
Model: strings.TrimSpace(base.Model),
|
||
Company: strings.TrimSpace(base.Company),
|
||
ExpectedProtocols: append([]string(nil), canonicalVehicleProtocols...),
|
||
ActualProtocols: []string{},
|
||
MissingProtocols: []string{},
|
||
ProtocolStatuses: []AccessProtocolStatus{},
|
||
ExpectationEvidence: "平台标准接入基线：GB32960 / JT808 / YUTONG_MQTT",
|
||
ConnectionState: "not_connected",
|
||
OnlineState: "never_reported",
|
||
FirstSeenEvidence: "尚未形成任何协议接入快照",
|
||
ReportIntervalProof: "需要连续上报样本后才能计算",
|
||
}
|
||
byProtocol := make(map[string]AccessEvidenceRow, len(items))
|
||
for _, item := range items {
|
||
protocol := strings.ToUpper(strings.TrimSpace(item.Protocol))
|
||
if protocol != "" {
|
||
byProtocol[protocol] = item
|
||
}
|
||
if row.Plate == "" {
|
||
row.Plate = strings.TrimSpace(item.Plate)
|
||
}
|
||
if row.OEM == "" {
|
||
row.OEM = strings.TrimSpace(item.OEM)
|
||
}
|
||
if row.Model == "" {
|
||
row.Model = strings.TrimSpace(item.Model)
|
||
}
|
||
if row.Company == "" {
|
||
row.Company = strings.TrimSpace(item.Company)
|
||
}
|
||
}
|
||
onlineCount := 0
|
||
unknownCount := 0
|
||
connectedCount := 0
|
||
for _, protocol := range canonicalVehicleProtocols {
|
||
item, connected := byProtocol[protocol]
|
||
if !connected {
|
||
threshold := accessProtocolThreshold(config, protocol)
|
||
row.MissingProtocols = append(row.MissingProtocols, protocol)
|
||
row.ProtocolStatuses = append(row.ProtocolStatuses, AccessProtocolStatus{
|
||
Protocol: protocol, Expected: true, Connected: false, OnlineState: "never_reported", ThresholdSec: threshold,
|
||
FirstSeenEvidence: "应接协议尚未形成实时快照", ReportIntervalEvidence: "尚无接收样本",
|
||
})
|
||
continue
|
||
}
|
||
source := buildAccessVehicleRow(item, config, now)
|
||
connectedCount++
|
||
row.ActualProtocols = append(row.ActualProtocols, protocol)
|
||
if source.OnlineState == "online" {
|
||
onlineCount++
|
||
}
|
||
if source.DelayAbnormal {
|
||
row.DelayAbnormal = true
|
||
}
|
||
row.ProtocolStatuses = append(row.ProtocolStatuses, AccessProtocolStatus{
|
||
Protocol: protocol, Expected: true, Connected: true, Provider: source.Provider,
|
||
FirstSeenAt: source.FirstSeenAt, LatestEventAt: source.LatestEventAt, LatestReceivedAt: source.LatestReceivedAt,
|
||
ReportIntervalSec: source.ReportIntervalSec, DataDelaySec: source.DataDelaySec, FreshnessSec: source.FreshnessSec,
|
||
OnlineState: source.OnlineState, ThresholdSec: source.ThresholdSec, DelayAbnormal: source.DelayAbnormal,
|
||
FirstSeenEvidence: source.FirstSeenEvidence, ReportIntervalEvidence: source.ReportIntervalProof,
|
||
Longitude: item.Longitude, Latitude: item.Latitude, SpeedKmh: item.SpeedKmh, SOCPercent: item.SOCPercent, TotalMileageKm: item.TotalMileageKm, DailyMileageKm: item.DailyMileageKm,
|
||
})
|
||
if row.FirstSeenAt == "" || source.FirstSeenAt != "" && source.FirstSeenAt < row.FirstSeenAt {
|
||
row.FirstSeenAt = source.FirstSeenAt
|
||
row.FirstSeenEvidence = source.FirstSeenEvidence
|
||
row.FirstSeenSource = source.FirstSeenSource
|
||
}
|
||
if row.LatestReceivedAt == "" || source.LatestReceivedAt > row.LatestReceivedAt {
|
||
copyAccessPrimaryFields(&row, source)
|
||
}
|
||
}
|
||
for protocol, item := range byProtocol {
|
||
if containsString(canonicalVehicleProtocols, protocol) {
|
||
continue
|
||
}
|
||
source := buildAccessVehicleRow(item, config, now)
|
||
row.ActualProtocols = append(row.ActualProtocols, protocol)
|
||
if source.OnlineState == "online" {
|
||
onlineCount++
|
||
}
|
||
if source.OnlineState == "unknown" {
|
||
unknownCount++
|
||
}
|
||
row.ProtocolStatuses = append(row.ProtocolStatuses, AccessProtocolStatus{
|
||
Protocol: protocol, Expected: false, Connected: true, Provider: source.Provider,
|
||
FirstSeenAt: source.FirstSeenAt, LatestEventAt: source.LatestEventAt, LatestReceivedAt: source.LatestReceivedAt,
|
||
ReportIntervalSec: source.ReportIntervalSec, DataDelaySec: source.DataDelaySec, FreshnessSec: source.FreshnessSec,
|
||
OnlineState: source.OnlineState, ThresholdSec: source.ThresholdSec, DelayAbnormal: source.DelayAbnormal,
|
||
FirstSeenEvidence: source.FirstSeenEvidence, ReportIntervalEvidence: source.ReportIntervalProof,
|
||
Longitude: item.Longitude, Latitude: item.Latitude, SpeedKmh: item.SpeedKmh, SOCPercent: item.SOCPercent, TotalMileageKm: item.TotalMileageKm, DailyMileageKm: item.DailyMileageKm,
|
||
})
|
||
if row.LatestReceivedAt == "" || source.LatestReceivedAt > row.LatestReceivedAt {
|
||
copyAccessPrimaryFields(&row, source)
|
||
}
|
||
}
|
||
switch {
|
||
case connectedCount == 0 && len(row.ActualProtocols) == 0:
|
||
row.ConnectionState = "not_connected"
|
||
row.OnlineState = "never_reported"
|
||
case connectedCount == len(canonicalVehicleProtocols) && onlineCount == connectedCount:
|
||
row.ConnectionState = "healthy"
|
||
row.OnlineState = "online"
|
||
case onlineCount == 0 && unknownCount > 0:
|
||
row.ConnectionState = "incomplete"
|
||
row.OnlineState = "unknown"
|
||
case onlineCount == 0:
|
||
row.ConnectionState = "offline"
|
||
row.OnlineState = "offline"
|
||
case connectedCount < len(canonicalVehicleProtocols):
|
||
row.ConnectionState = "incomplete"
|
||
row.OnlineState = "online"
|
||
default:
|
||
row.ConnectionState = "degraded"
|
||
row.OnlineState = "online"
|
||
}
|
||
return row
|
||
}
|
||
|
||
func accessProtocolThreshold(config AccessThresholdConfig, protocol string) int {
|
||
for _, override := range config.Protocols {
|
||
if strings.EqualFold(override.Protocol, protocol) {
|
||
return override.ThresholdSec
|
||
}
|
||
}
|
||
return config.DefaultThresholdSec
|
||
}
|
||
|
||
func copyAccessPrimaryFields(target *AccessVehicleRow, source AccessVehicleRow) {
|
||
target.Protocol = source.Protocol
|
||
target.Provider = source.Provider
|
||
target.Source = source.Source
|
||
target.LatestEventAt = source.LatestEventAt
|
||
target.LatestReceivedAt = source.LatestReceivedAt
|
||
target.ReportIntervalSec = source.ReportIntervalSec
|
||
target.DataDelaySec = source.DataDelaySec
|
||
target.FreshnessSec = source.FreshnessSec
|
||
target.ThresholdSec = source.ThresholdSec
|
||
target.LatestMessageType = source.LatestMessageType
|
||
target.LatestEventID = source.LatestEventID
|
||
target.LatestError = source.LatestError
|
||
target.ReportIntervalProof = source.ReportIntervalProof
|
||
target.ReportSampleCount = source.ReportSampleCount
|
||
}
|
||
|
||
func buildAccessVehicleRow(item AccessEvidenceRow, config AccessThresholdConfig, now time.Time) AccessVehicleRow {
|
||
threshold := config.DefaultThresholdSec
|
||
for _, override := range config.Protocols {
|
||
if strings.EqualFold(strings.TrimSpace(override.Protocol), strings.TrimSpace(item.Protocol)) {
|
||
threshold = override.ThresholdSec
|
||
break
|
||
}
|
||
}
|
||
row := AccessVehicleRow{
|
||
VIN: strings.TrimSpace(item.VIN),
|
||
Plate: strings.TrimSpace(item.Plate),
|
||
OEM: strings.TrimSpace(item.OEM),
|
||
Model: strings.TrimSpace(item.Model),
|
||
Company: strings.TrimSpace(item.Company),
|
||
Protocol: strings.TrimSpace(item.Protocol),
|
||
Provider: strings.TrimSpace(item.Provider),
|
||
Source: strings.TrimSpace(item.Source),
|
||
FirstSeenAt: normalizeAccessTime(item.FirstSeenAt),
|
||
LatestEventAt: normalizeAccessTime(item.LatestEventAt),
|
||
LatestReceivedAt: normalizeAccessTime(item.LatestReceivedAt),
|
||
ReportIntervalSec: item.ReportIntervalSec,
|
||
ThresholdSec: threshold,
|
||
LatestMessageType: firstNonEmpty(strings.TrimSpace(item.LatestMessageType), accessMessageType(item.Protocol)),
|
||
LatestEventID: strings.TrimSpace(item.LatestEventID),
|
||
LatestError: strings.TrimSpace(item.LatestError),
|
||
FirstSeenEvidence: strings.TrimSpace(item.FirstSeenEvidence),
|
||
FirstSeenSource: strings.TrimSpace(item.FirstSeenSource),
|
||
ReportIntervalProof: strings.TrimSpace(item.ReportIntervalProof),
|
||
ReportSampleCount: item.ReportSampleCount,
|
||
}
|
||
if row.FirstSeenEvidence == "" {
|
||
row.FirstSeenEvidence = "现有实时快照不保存首次接入时间"
|
||
}
|
||
if row.ReportIntervalProof == "" {
|
||
row.ReportIntervalProof = "需要连续上报样本后才能计算"
|
||
}
|
||
eventAt, eventOK := parseAccessTime(item.LatestEventAt)
|
||
receivedAt, receivedOK := parseAccessTime(item.LatestReceivedAt)
|
||
if eventOK && receivedOK {
|
||
delay := int(receivedAt.Sub(eventAt).Seconds())
|
||
row.DataDelaySec = &delay
|
||
row.DelayAbnormal = delay < 0 || delay > config.DelayThresholdSec
|
||
if delay < 0 && row.LatestError == "" {
|
||
row.LatestError = "接收时间早于事件时间"
|
||
}
|
||
}
|
||
latest, latestOK := receivedAt, receivedOK
|
||
if !latestOK {
|
||
latest, latestOK = parseAccessTime(item.LatestUpdatedAt)
|
||
}
|
||
switch {
|
||
case strings.TrimSpace(item.Protocol) == "" && !latestOK && !eventOK:
|
||
row.OnlineState = "never_reported"
|
||
case !latestOK:
|
||
row.OnlineState = "unknown"
|
||
if row.LatestError == "" {
|
||
row.LatestError = "缺少可解析的接收时间"
|
||
}
|
||
default:
|
||
freshness := int(now.Sub(latest).Seconds())
|
||
if freshness < 0 {
|
||
freshness = 0
|
||
}
|
||
row.FreshnessSec = &freshness
|
||
if freshness <= threshold {
|
||
row.OnlineState = "online"
|
||
} else {
|
||
row.OnlineState = "offline"
|
||
}
|
||
}
|
||
return row
|
||
}
|
||
|
||
func keepAccessRow(row AccessVehicleRow, query AccessQuery) bool {
|
||
keyword := strings.ToLower(strings.TrimSpace(query.Keyword))
|
||
if keyword != "" && !strings.Contains(strings.ToLower(row.VIN), keyword) && !strings.Contains(strings.ToLower(row.Plate), keyword) && !strings.Contains(strings.ToLower(row.OEM), keyword) && !strings.Contains(strings.ToLower(row.Model), keyword) && !strings.Contains(strings.ToLower(row.Company), keyword) {
|
||
return false
|
||
}
|
||
if value := strings.TrimSpace(query.Protocol); value != "" && !containsString(row.ActualProtocols, value) {
|
||
return false
|
||
}
|
||
if value := strings.TrimSpace(query.OEM); value != "" && !strings.EqualFold(value, row.OEM) {
|
||
return false
|
||
}
|
||
if value := strings.ToLower(strings.TrimSpace(query.Model)); value != "" && !strings.Contains(strings.ToLower(row.Model), value) {
|
||
return false
|
||
}
|
||
if value := strings.ToLower(strings.TrimSpace(query.Provider)); value != "" {
|
||
matched := false
|
||
for _, status := range row.ProtocolStatuses {
|
||
if strings.Contains(strings.ToLower(status.Provider), value) {
|
||
matched = true
|
||
break
|
||
}
|
||
}
|
||
if !matched {
|
||
return false
|
||
}
|
||
}
|
||
if !accessTimeMatches(row.FirstSeenAt, query.FirstSeenFrom, query.FirstSeenTo) || !accessTimeMatches(row.LatestReceivedAt, query.LatestSeenFrom, query.LatestSeenTo) {
|
||
return false
|
||
}
|
||
if value := strings.TrimSpace(query.OnlineState); value != "" && value != "all" && value != row.OnlineState {
|
||
return false
|
||
}
|
||
if value := strings.TrimSpace(query.ConnectionState); value != "" && value != "all" {
|
||
if value == "attention" {
|
||
if row.ConnectionState == "healthy" {
|
||
return false
|
||
}
|
||
} else if value != row.ConnectionState {
|
||
return false
|
||
}
|
||
}
|
||
switch strings.TrimSpace(query.DelayState) {
|
||
case "abnormal":
|
||
return row.DelayAbnormal
|
||
case "normal":
|
||
return row.DataDelaySec != nil && !row.DelayAbnormal
|
||
}
|
||
return true
|
||
}
|
||
|
||
func validateAccessQuery(query AccessQuery) error {
|
||
for _, item := range []struct{ name, from, to string }{{"首次接入", query.FirstSeenFrom, query.FirstSeenTo}, {"最新上报", query.LatestSeenFrom, query.LatestSeenTo}} {
|
||
start, startOK := parseAccessFilterTime(item.from)
|
||
end, endOK := parseAccessFilterTime(item.to)
|
||
if strings.TrimSpace(item.from) != "" && !startOK || strings.TrimSpace(item.to) != "" && !endOK {
|
||
return clientError{Code: "ACCESS_TIME_INVALID", Message: item.name + "时间格式无效"}
|
||
}
|
||
if startOK && endOK && start.After(end) {
|
||
return clientError{Code: "ACCESS_TIME_RANGE_INVALID", Message: item.name + "开始时间不能晚于结束时间"}
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func parseAccessFilterTime(value string) (time.Time, bool) {
|
||
if strings.TrimSpace(value) == "" {
|
||
return time.Time{}, false
|
||
}
|
||
if parsed, ok := parseTrackRequestTime(value); ok {
|
||
return parsed, true
|
||
}
|
||
return parseAccessTime(value)
|
||
}
|
||
|
||
func accessTimeMatches(value, from, to string) bool {
|
||
if strings.TrimSpace(from) == "" && strings.TrimSpace(to) == "" {
|
||
return true
|
||
}
|
||
actual, ok := parseAccessTime(value)
|
||
if !ok {
|
||
return false
|
||
}
|
||
if start, ok := parseAccessFilterTime(from); ok && actual.Before(start) {
|
||
return false
|
||
}
|
||
if end, ok := parseAccessFilterTime(to); ok && actual.After(end) {
|
||
return false
|
||
}
|
||
return true
|
||
}
|
||
|
||
func validateAccessThresholdUpdate(update AccessThresholdUpdate) error {
|
||
if update.Version <= 0 {
|
||
return clientError{Code: "ACCESS_THRESHOLD_VERSION_REQUIRED", Message: "阈值版本不能为空"}
|
||
}
|
||
if update.DefaultThresholdSec < 30 || update.DefaultThresholdSec > 86400 {
|
||
return clientError{Code: "ACCESS_THRESHOLD_INVALID", Message: "全局在线阈值必须在 30 秒到 24 小时之间"}
|
||
}
|
||
if update.DelayThresholdSec < 1 || update.DelayThresholdSec > 3600 {
|
||
return clientError{Code: "ACCESS_DELAY_THRESHOLD_INVALID", Message: "延迟阈值必须在 1 秒到 1 小时之间"}
|
||
}
|
||
if update.LongOfflineSec < update.DefaultThresholdSec || update.LongOfflineSec > 604800 {
|
||
return clientError{Code: "ACCESS_LONG_OFFLINE_INVALID", Message: "长离线阈值必须不小于在线阈值且不超过 7 天"}
|
||
}
|
||
seen := map[string]struct{}{}
|
||
for _, item := range update.Protocols {
|
||
protocol := strings.ToUpper(strings.TrimSpace(item.Protocol))
|
||
if protocol == "" || item.ThresholdSec < 30 || item.ThresholdSec > 86400 {
|
||
return clientError{Code: "ACCESS_PROTOCOL_THRESHOLD_INVALID", Message: "协议阈值必须包含协议名且位于 30 秒到 24 小时之间"}
|
||
}
|
||
if _, exists := seen[protocol]; exists {
|
||
return clientError{Code: "ACCESS_PROTOCOL_THRESHOLD_DUPLICATED", Message: "协议阈值不能重复"}
|
||
}
|
||
seen[protocol] = struct{}{}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func normalizeAccessThresholdConfig(config AccessThresholdConfig, now time.Time) AccessThresholdConfig {
|
||
defaults := defaultAccessThresholds(now)
|
||
if config.Version <= 0 {
|
||
return defaults
|
||
}
|
||
if config.DefaultThresholdSec <= 0 {
|
||
config.DefaultThresholdSec = defaults.DefaultThresholdSec
|
||
}
|
||
if config.DelayThresholdSec <= 0 {
|
||
config.DelayThresholdSec = defaults.DelayThresholdSec
|
||
}
|
||
if config.LongOfflineSec <= 0 {
|
||
config.LongOfflineSec = defaults.LongOfflineSec
|
||
}
|
||
if config.Protocols == nil {
|
||
config.Protocols = []AccessProtocolThreshold{}
|
||
}
|
||
if config.Audit == nil {
|
||
config.Audit = []AccessThresholdAudit{}
|
||
}
|
||
return config
|
||
}
|
||
|
||
func parseAccessTime(value string) (time.Time, bool) {
|
||
value = strings.TrimSpace(value)
|
||
if value == "" {
|
||
return time.Time{}, false
|
||
}
|
||
for _, layout := range []string{time.RFC3339Nano, time.RFC3339} {
|
||
if parsed, err := time.Parse(layout, value); err == nil {
|
||
return parsed, true
|
||
}
|
||
}
|
||
for _, layout := range []string{"2006-01-02 15:04:05.000", "2006-01-02 15:04:05"} {
|
||
if parsed, err := time.ParseInLocation(layout, value, time.Local); err == nil {
|
||
return parsed, true
|
||
}
|
||
}
|
||
return time.Time{}, false
|
||
}
|
||
|
||
func normalizeAccessTime(value string) string {
|
||
parsed, ok := parseAccessTime(value)
|
||
if !ok {
|
||
return ""
|
||
}
|
||
return parsed.Format(time.RFC3339)
|
||
}
|
||
|
||
func accessMessageType(protocol string) string {
|
||
switch strings.ToUpper(strings.TrimSpace(protocol)) {
|
||
case "GB32960":
|
||
return "实时信息上报"
|
||
case "JT808":
|
||
return "位置信息汇报"
|
||
case "YUTONG_MQTT":
|
||
return "实时遥测"
|
||
default:
|
||
return ""
|
||
}
|
||
}
|
||
|
||
func accessStateRank(state string) int {
|
||
switch state {
|
||
case "online":
|
||
return 0
|
||
case "offline":
|
||
return 1
|
||
case "never_reported":
|
||
return 2
|
||
default:
|
||
return 3
|
||
}
|
||
}
|
||
|
||
func reportedOnDay(value string, day time.Time) bool {
|
||
parsed, ok := parseAccessTime(value)
|
||
if !ok {
|
||
return false
|
||
}
|
||
year, month, date := parsed.In(day.Location()).Date()
|
||
wantYear, wantMonth, wantDate := day.Date()
|
||
return year == wantYear && month == wantMonth && date == wantDate
|
||
}
|
||
|
||
func addAccessDistribution(items map[string]*AccessDistribution, name string, online bool) {
|
||
item := items[name]
|
||
if item == nil {
|
||
item = &AccessDistribution{Name: name}
|
||
items[name] = item
|
||
}
|
||
item.Total++
|
||
if online {
|
||
item.Online++
|
||
}
|
||
}
|
||
|
||
func sortedAccessDistributions(items map[string]*AccessDistribution) []AccessDistribution {
|
||
result := make([]AccessDistribution, 0, len(items))
|
||
for _, item := range items {
|
||
value := *item
|
||
if value.Total > 0 {
|
||
value.OnlineRate = float64(value.Online) / float64(value.Total) * 100
|
||
}
|
||
result = append(result, value)
|
||
}
|
||
sort.Slice(result, func(i, j int) bool {
|
||
if result[i].Total != result[j].Total {
|
||
return result[i].Total > result[j].Total
|
||
}
|
||
return result[i].Name < result[j].Name
|
||
})
|
||
return result
|
||
}
|