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: []string{}, ActualProtocols: []string{}, MissingProtocols: []string{}, ProtocolStatuses: []AccessProtocolStatus{}, MasterDataIssues: []string{}, ExpectationEvidence: "尚未接入业务系统的应接协议口径;当前只展示真实接入来源,不推断车辆必须具备三协议", 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 for _, protocol := range canonicalVehicleProtocols { item, connected := byProtocol[protocol] if !connected { threshold := accessProtocolThreshold(config, protocol) row.ProtocolStatuses = append(row.ProtocolStatuses, AccessProtocolStatus{ Protocol: protocol, Expected: false, Connected: false, OnlineState: "never_reported", ThresholdSec: threshold, FirstSeenEvidence: "当前未发现该协议来源;在业务应接口径接入前不判定为缺失", ReportIntervalEvidence: "尚无接收样本", }) continue } source := buildAccessVehicleRow(item, config, now) row.ActualProtocols = append(row.ActualProtocols, protocol) if source.OnlineState == "online" { onlineCount++ } if source.OnlineState == "unknown" { unknownCount++ } if source.DelayAbnormal { row.DelayAbnormal = true } 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.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) } } extraProtocols := make([]string, 0) for protocol := range byProtocol { if containsString(canonicalVehicleProtocols, protocol) { continue } extraProtocols = append(extraProtocols, protocol) } sort.Strings(extraProtocols) for _, protocol := range extraProtocols { item := byProtocol[protocol] 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) } } if accessMasterDataMissing(row.OEM) { row.MasterDataIssues = append(row.MasterDataIssues, "车辆品牌未维护") } for _, status := range row.ProtocolStatuses { if status.Connected && accessMasterDataMissing(status.Provider) { row.MasterDataIssues = append(row.MasterDataIssues, status.Protocol+" 接入方未维护") } } sourceCount := len(row.ActualProtocols) switch { case sourceCount == 0: row.ConnectionState = "not_connected" row.OnlineState = "never_reported" case onlineCount == 0 && unknownCount == sourceCount: row.ConnectionState = "degraded" row.OnlineState = "unknown" case onlineCount == 0: row.ConnectionState = "offline" row.OnlineState = "offline" case onlineCount < sourceCount || row.DelayAbnormal: row.ConnectionState = "degraded" row.OnlineState = "online" case len(row.MasterDataIssues) > 0: row.ConnectionState = "incomplete" row.OnlineState = "online" default: row.ConnectionState = "healthy" row.OnlineState = "online" } return row } func accessMasterDataMissing(value string) bool { switch strings.ToLower(strings.TrimSpace(value)) { case "", "未维护", "待维护", "unknown", "未知": return true default: return false } } 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 != "" { if value == "未维护" { if !accessMasterDataMissing(row.OEM) { return false } } else if !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 }