package platform import ( "bufio" "context" "crypto/rand" "encoding/csv" "encoding/hex" "encoding/json" "errors" "fmt" "math" "net/url" "os" "path/filepath" "sort" "strconv" "strings" "sync" "time" ) type Store interface { DashboardSummary(context.Context) (DashboardSummary, error) Vehicles(context.Context, url.Values) (Page[VehicleRow], error) VehicleCoverage(context.Context, url.Values) (Page[VehicleCoverageRow], error) VehicleCoverageSummary(context.Context, url.Values) (VehicleCoverageSummary, error) VehicleServiceSummary(context.Context) (VehicleServiceSummary, error) VehicleRealtime(context.Context, url.Values) (Page[VehicleRealtimeRow], error) RealtimeLocations(context.Context, url.Values) (Page[RealtimeLocationRow], error) HistoryLocations(context.Context, url.Values) (Page[HistoryLocationRow], error) HistoryLocationsFromTDengine(context.Context, url.Values) (Page[HistoryLocationRow], error) RawFrames(context.Context, RawFrameQuery) (Page[RawFrameRow], error) MileageSummary(context.Context, url.Values) (MileageSummary, error) DailyMileage(context.Context, url.Values) (Page[DailyMileageRow], error) MileageStatistics(context.Context, url.Values) (MileageStatistics, error) QualitySummary(context.Context, url.Values) (QualitySummary, error) QualityIssues(context.Context, url.Values) (Page[QualityIssueRow], error) OpsHealth(context.Context) (OpsHealth, error) } type VehicleOverviewBatchStore interface { VehicleServiceOverviews(context.Context, VehicleOverviewBatchQuery) (Page[VehicleServiceOverview], error) } type VehicleProfileStore interface { VehicleProfile(context.Context, string) (VehicleProfile, bool, error) SaveVehicleProfile(context.Context, string, VehicleProfileInput) (VehicleProfile, error) SyncVehicleProfiles(context.Context, VehicleProfileSyncRequest) (VehicleProfileSyncResult, error) } type RawFrameQuery struct { Protocol string `json:"protocol"` VIN string `json:"vin"` Keyword string `json:"keyword"` DateFrom string `json:"dateFrom"` DateTo string `json:"dateTo"` Fields []string `json:"fields"` IncludeFields bool `json:"includeFields"` Limit int `json:"limit"` Offset int `json:"offset"` SkipCount bool `json:"-"` } type VehicleOverviewBatchQuery struct { Keywords []string `json:"keywords"` Protocol string `json:"protocol"` Limit int `json:"limit"` Offset int `json:"offset"` } type Service struct { store Store runtime RuntimeInfo exportsMu sync.RWMutex exports map[string]*HistoryExportJob exportDir string exportSlots chan struct{} } type clientError struct { Code string Message string } func (e clientError) Error() string { return e.Message } func asClientError(err error) (clientError, bool) { var target clientError if errors.As(err, &target) { return target, true } return clientError{}, false } var canonicalVehicleProtocols = []string{"GB32960", "JT808", "YUTONG_MQTT"} var qualityAlertRuleTemplates = []QualityAlertRule{ { IssueType: "NO_SOURCE", Title: "无数据来源", Level: "P0", Owner: "平台接入", Trigger: "绑定车辆连续无 GB32960、JT808、MQTT 来源证据", Notify: "立即通知接入运维,30 分钟未恢复升级给业务责任人", SLA: "30 分钟确认", }, { IssueType: "VIN_MISSING", Title: "VIN 缺失", Level: "P0", Owner: "车辆档案", Trigger: "来源有数据但无法归并到 VIN", Notify: "通知档案维护人补齐车牌、手机号和 VIN 映射", SLA: "2 小时修复", }, { IssueType: "LINK_GAP", Title: "链路中断", Level: "P1", Owner: "链路运维", Trigger: "车辆或平台来源上报间断、Redis 在线状态过期", Notify: "通知平台转发和网关值班人,持续异常进入日报", SLA: "1 小时恢复", }, { IssueType: "FIELD_MISSING", Title: "字段缺失", Level: "P1", Owner: "协议解析", Trigger: "核心字段缺失导致定位、里程或统计不可用", Notify: "通知解析负责人核对字段映射和原始记录样本", SLA: "当日闭环", }, } var qualityNotificationPolicies = []QualityNotificationPolicy{ { Name: "P0 实时中断", Target: "接入运维 + 业务责任人", Channel: "站内告警 / 邮件 / 企业微信", Condition: "无来源、VIN 缺失、存储不可写", EscalationMinutes: 30, AcceptanceCriteria: "来源恢复并持续 10 分钟,车辆服务可查到实时与历史证据", }, { Name: "P1 数据质量", Target: "协议解析 + 数据治理", Channel: "站内告警 / 每日汇总邮件", Condition: "字段缺失、链路间断、容量风险", EscalationMinutes: 120, AcceptanceCriteria: "核心字段恢复解析,影响车辆可通过原始记录与历史证据复核", }, { Name: "P2 趋势关注", Target: "数智中心", Channel: "周报 / 趋势看板", Condition: "单源车辆、档案缺项、来源覆盖下降", EscalationMinutes: 1440, AcceptanceCriteria: "趋势原因已归档,责任人和后续治理动作明确", }, } var qualityIssuePriorityWeight = map[string]int{ "NO_SOURCE": 10, "VIN_MISSING": 9, "LINK_GAP": 8, "FIELD_MISSING": 7, } func completeProtocolStats(stats []ProtocolStat) []ProtocolStat { byProtocol := make(map[string]ProtocolStat, len(stats)+len(canonicalVehicleProtocols)) for _, stat := range stats { protocol := strings.TrimSpace(stat.Protocol) if protocol == "" { continue } stat.Protocol = protocol byProtocol[protocol] = stat } out := make([]ProtocolStat, 0, len(byProtocol)+len(canonicalVehicleProtocols)) seen := map[string]struct{}{} for _, protocol := range canonicalVehicleProtocols { stat := byProtocol[protocol] stat.Protocol = protocol out = append(out, stat) seen[protocol] = struct{}{} } extra := make([]string, 0) for protocol := range byProtocol { if _, ok := seen[protocol]; !ok { extra = append(extra, protocol) } } sort.Strings(extra) for _, protocol := range extra { out = append(out, byProtocol[protocol]) } return out } func buildSourceReadinessPlan(summary VehicleServiceSummary, health OpsHealth, quality QualitySummary) SourceReadinessPlan { missingByProtocol := make(map[string]int, len(summary.MissingSources)) for _, item := range summary.MissingSources { missingByProtocol[strings.TrimSpace(item.Protocol)] = item.Count } issueByType := make(map[string]int, len(quality.IssueTypes)) for _, item := range quality.IssueTypes { issueByType[strings.TrimSpace(item.Name)] = item.Count } protocolStats := completeProtocolStats(summary.Protocols) rows := make([]SourceReadinessRow, 0, len(protocolStats)) for _, stat := range protocolStats { protocol := strings.TrimSpace(stat.Protocol) if protocol == "" { continue } missing := missingByProtocol[protocol] row := SourceReadinessRow{ Protocol: protocol, Role: sourceReadinessRole(protocol), Online: stat.Online, Total: stat.Total, OnlineRate: sourceOnlineRate(stat), MissingVehicles: missing, Severity: "ok", Status: "生产可用", Evidence: sourceReadinessEvidence(stat, missing, health), Action: "保持实时、轨迹、原始记录和统计链路巡检。", Acceptance: "该来源有在线车辆、无消息积压,车辆服务可回查实时、轨迹和原始记录。", VehiclesHash: buildHash("vehicles", "", protocol, map[string]string{"missingProtocol": protocol}), RealtimeHash: buildHash("realtime", "", protocol, map[string]string{"online": "online"}), HistoryHash: buildHash("history-query", "", protocol, map[string]string{"tab": "raw", "includeFields": "true"}), AlertHash: buildHash("alert-events", "", protocol, nil), } if stat.Total <= 0 { row.Severity = "error" row.Status = "未形成来源证据" row.Action = "确认平台账号、端口监听、订阅配置和协议接入服务是否启动。" row.Acceptance = "该来源至少出现绑定车辆和实时原始记录。" } else if stat.Online <= 0 { row.Severity = "error" row.Status = "来源全离线" row.Action = "优先检查上游平台转发、网关连接、登录/鉴权回复和 Redis 在线 TTL。" row.Acceptance = "该协议恢复在线车辆,Redis 在线状态持续续期。" } else if missing > summary.TotalVehicles/2 && summary.TotalVehicles > 0 { row.Severity = "warning" row.Status = "覆盖不足" row.Action = "核对该协议车辆范围、绑定表和平台转发清单,确认缺失车辆是否应接入。" row.Acceptance = "缺失来源车辆下降,单源车辆可解释。" } if health.KafkaLag != nil && *health.KafkaLag > 0 { row.Severity = maxSourceSeverity(row.Severity, "warning") row.Status = row.Status + " / Kafka积压" row.Action = row.Action + " 同时检查 Kafka 消费 Lag,避免历史和统计滞后。" } if issueByType["NO_SOURCE"] > 0 && stat.Total <= 0 { row.Severity = "error" } rows = append(rows, row) } return SourceReadinessPlan{ TotalVehicles: summary.TotalVehicles, BoundVehicles: summary.BoundVehicles, IdentityRequiredVehicles: summary.IdentityRequiredVehicles, OnlineVehicles: summary.OnlineVehicles, KafkaLag: health.KafkaLag, ActiveConnections: health.ActiveConnections, RedisOnlineKeys: health.RedisOnlineKeys, PlatformRelease: health.Runtime.PlatformRelease, Sources: rows, } } func sourceReadinessRole(protocol string) string { switch strings.ToUpper(strings.TrimSpace(protocol)) { case "GB32960": return "整车与氢能实时数据主来源" case "JT808": return "位置、总里程和设备在线主来源" case "YUTONG_MQTT": return "宇通派生工况与燃料电池补充来源" default: return "扩展接入来源" } } func sourceOnlineRate(stat ProtocolStat) float64 { if stat.Total <= 0 { return 0 } return float64(stat.Online) / float64(stat.Total) * 100 } func sourceReadinessEvidence(stat ProtocolStat, missing int, health OpsHealth) string { parts := []string{ "在线 " + strconv.Itoa(stat.Online) + "/" + strconv.Itoa(stat.Total), "缺失车辆 " + strconv.Itoa(missing), } if health.KafkaLag != nil { parts = append(parts, "Kafka Lag "+strconv.Itoa(*health.KafkaLag)) } if health.RedisOnlineKeys != nil { parts = append(parts, "Redis在线Key "+strconv.Itoa(*health.RedisOnlineKeys)) } return strings.Join(parts, ",") } func maxSourceSeverity(left string, right string) string { weight := map[string]int{"ok": 0, "warning": 1, "error": 2} if weight[right] > weight[left] { return right } return left } func missingCanonicalProtocols(protocols []string) []string { missing := make([]string, 0, len(canonicalVehicleProtocols)) for _, protocol := range canonicalVehicleProtocols { if !containsString(protocols, protocol) { missing = append(missing, protocol) } } return missing } func buildVehicleCoverageSourceStatus(protocols []string, onlineProtocols []string, lastSeen string) []VehicleSourceStatus { statuses := make([]VehicleSourceStatus, 0, len(protocols)) for _, protocol := range protocols { protocol = strings.TrimSpace(protocol) if protocol == "" { continue } statuses = append(statuses, VehicleSourceStatus{ Protocol: protocol, Online: containsString(onlineProtocols, protocol), LastSeen: lastSeen, HasRealtime: true, }) } return completeVehicleSourceStatus(statuses, "") } func NewService(store Store) *Service { return newService(store, RuntimeInfo{}) } func NewServiceWithRuntime(store Store, runtime RuntimeInfo) *Service { return newService(store, runtime) } func newService(store Store, runtime RuntimeInfo) *Service { exportDir := strings.TrimSpace(runtime.ExportDir) if exportDir == "" { exportDir = filepath.Join(os.TempDir(), "lingniu-vehicle-platform-exports") } service := &Service{store: store, runtime: runtime, exports: map[string]*HistoryExportJob{}, exportDir: exportDir, exportSlots: make(chan struct{}, 1)} service.loadHistoryExports() return service } func (s *Service) DashboardSummary(ctx context.Context) (DashboardSummary, error) { return s.store.DashboardSummary(ctx) } const ( monitorVehicleLimit = 10000 monitorPointLimit = 2000 monitorPointZoom = 11 monitorClusterGridPixels = 64 vehicleSearchKeywordLimit = 100 ) type monitorBounds struct { minLongitude float64 minLatitude float64 maxLongitude float64 maxLatitude float64 } type monitorAlertStore interface { ActiveAlertVINs(context.Context, string) ([]string, error) } func parseMonitorBounds(value string) (monitorBounds, bool, error) { value = strings.TrimSpace(value) if value == "" { return monitorBounds{}, false, nil } parts := strings.Split(value, ",") if len(parts) != 4 { return monitorBounds{}, false, clientError{Code: "MONITOR_BOUNDS_INVALID", Message: "地图视口必须为最小经度、最小纬度、最大经度、最大纬度"} } values := [4]float64{} for index, part := range parts { parsed, err := strconv.ParseFloat(strings.TrimSpace(part), 64) if err != nil || math.IsNaN(parsed) || math.IsInf(parsed, 0) { return monitorBounds{}, false, clientError{Code: "MONITOR_BOUNDS_INVALID", Message: "地图视口坐标无效"} } values[index] = parsed } if values[0] < -180 || values[2] > 180 || values[1] < -90 || values[3] > 90 || values[0] >= values[2] || values[1] >= values[3] { return monitorBounds{}, false, clientError{Code: "MONITOR_BOUNDS_INVALID", Message: "地图视口范围无效"} } return monitorBounds{minLongitude: values[0], minLatitude: values[1], maxLongitude: values[2], maxLatitude: values[3]}, true, nil } func (bounds monitorBounds) contains(longitude, latitude float64) bool { return longitude >= bounds.minLongitude && longitude <= bounds.maxLongitude && latitude >= bounds.minLatitude && latitude <= bounds.maxLatitude } func monitorVehicleStatus(row VehicleRealtimeRow) string { if strings.TrimSpace(row.LastSeen) == "" { return "unknown" } if !row.Online { return "offline" } if row.SpeedKmh > 3 { return "driving" } return "idle" } func normalizeMonitorQuery(query url.Values) url.Values { next := make(url.Values, len(query)+2) for key, values := range query { next[key] = append([]string(nil), values...) } next.Set("limit", strconv.Itoa(monitorVehicleLimit)) next.Set("offset", "0") if next.Get("vin") == "" && strings.TrimSpace(next.Get("keywords")) == "" && strings.TrimSpace(next.Get("keyword")) != "" { next.Set("vin", strings.TrimSpace(next.Get("keyword"))) } switch strings.TrimSpace(next.Get("status")) { case "online": next.Set("online", "online") case "offline": next.Set("online", "offline") } return next } func vehicleSearchKeywords(query url.Values) []string { rawValues := query["keywords"] if len(rawValues) == 0 { rawValues = []string{firstNonEmpty(query.Get("vin"), query.Get("keyword"))} } result := make([]string, 0, len(rawValues)) seen := map[string]struct{}{} for _, raw := range rawValues { parts := strings.FieldsFunc(raw, func(char rune) bool { switch char { case ',', ',', '、', ';', ';', '\n', '\r', '\t', ' ': return true default: return false } }) for _, part := range parts { part = strings.TrimSpace(part) if part == "" { continue } key := strings.ToLower(part) if _, exists := seen[key]; exists { continue } seen[key] = struct{}{} result = append(result, part) if len(result) == vehicleSearchKeywordLimit { return result } } } return result } func matchesMonitorStatus(row VehicleRealtimeRow, requested string) bool { requested = strings.TrimSpace(requested) return requested == "" || monitorVehicleStatus(row) == requested || requested == "online" && row.Online } func (s *Service) MonitorSummary(ctx context.Context, query url.Values) (MonitorSummary, error) { vehicles, err := s.store.VehicleRealtime(ctx, normalizeMonitorQuery(query)) if err != nil { return MonitorSummary{}, err } return s.buildMonitorSummary(ctx, query, vehicles) } func (s *Service) buildMonitorSummary(ctx context.Context, query url.Values, vehicles Page[VehicleRealtimeRow]) (MonitorSummary, error) { dashboard, err := s.store.DashboardSummary(ctx) if err != nil { return MonitorSummary{}, err } result := MonitorSummary{ ActiveToday: dashboard.ActiveToday, FrameToday: dashboard.FrameToday, Truncated: vehicles.Total > len(vehicles.Items), AsOf: time.Now().UTC().Format(time.RFC3339), } activeAlertVINs := map[string]struct{}{} if store, ok := s.store.(monitorAlertStore); ok { vins, alertErr := store.ActiveAlertVINs(ctx, strings.TrimSpace(query.Get("protocol"))) if alertErr != nil { return MonitorSummary{}, alertErr } for _, vin := range vins { activeAlertVINs[strings.TrimSpace(vin)] = struct{}{} } result.AlertDataAvailable = true } for _, vehicle := range vehicles.Items { if !matchesMonitorStatus(vehicle, query.Get("status")) { continue } result.TotalVehicles++ if _, active := activeAlertVINs[vehicle.VIN]; active { result.AlertVehicles++ } switch monitorVehicleStatus(vehicle) { case "driving": result.DrivingVehicles++ result.OnlineVehicles++ case "idle": result.IdleVehicles++ result.OnlineVehicles++ case "offline": result.OfflineVehicles++ default: result.UnknownVehicles++ } } return result, nil } func (s *Service) MonitorWorkspace(ctx context.Context, query url.Values) (MonitorWorkspaceResponse, error) { vehicles, err := s.store.VehicleRealtime(ctx, normalizeMonitorQuery(query)) if err != nil { return MonitorWorkspaceResponse{}, err } summary, err := s.buildMonitorSummary(ctx, query, vehicles) if err != nil { return MonitorWorkspaceResponse{}, err } monitorMap, err := buildMonitorMapResponse(vehicles, query) if err != nil { return MonitorWorkspaceResponse{}, err } railLimit := parsePositive(query.Get("railLimit"), 200) if railLimit > 200 { railLimit = 200 } if railLimit > len(vehicles.Items) { railLimit = len(vehicles.Items) } railItems := append([]VehicleRealtimeRow(nil), vehicles.Items[:railLimit]...) return MonitorWorkspaceResponse{ Summary: summary, Vehicles: Page[VehicleRealtimeRow]{Items: railItems, Total: vehicles.Total, Limit: railLimit, Offset: 0}, Map: monitorMap, }, nil } func (s *Service) MonitorMap(ctx context.Context, query url.Values) (MonitorMapResponse, error) { vehicles, err := s.store.VehicleRealtime(ctx, normalizeMonitorQuery(query)) if err != nil { return MonitorMapResponse{}, err } return buildMonitorMapResponse(vehicles, query) } func buildMonitorMapResponse(vehicles Page[VehicleRealtimeRow], query url.Values) (MonitorMapResponse, error) { zoom := parsePositive(query.Get("zoom"), 5) if zoom > 20 { zoom = 20 } bounds, hasBounds, err := parseMonitorBounds(query.Get("bounds")) if err != nil { return MonitorMapResponse{}, err } result := MonitorMapResponse{ Mode: "clusters", Zoom: zoom, Truncated: vehicles.Total > len(vehicles.Items), Points: []MonitorMapPoint{}, Clusters: []MonitorMapCluster{}, AsOf: time.Now().UTC().Format(time.RFC3339), } points := make([]MonitorMapPoint, 0, len(vehicles.Items)) for _, vehicle := range vehicles.Items { if !matchesMonitorStatus(vehicle, query.Get("status")) { continue } result.Total++ if vehicle.Longitude < 73 || vehicle.Longitude > 135 || vehicle.Latitude < 18 || vehicle.Latitude > 54 { continue } if hasBounds && !bounds.contains(vehicle.Longitude, vehicle.Latitude) { continue } points = append(points, MonitorMapPoint{ VIN: vehicle.VIN, Plate: vehicle.Plate, Protocol: vehicle.PrimaryProtocol, Protocols: vehicle.Protocols, Longitude: vehicle.Longitude, Latitude: vehicle.Latitude, SpeedKmh: vehicle.SpeedKmh, SOCPercent: vehicle.SOCPercent, TotalMileageKm: vehicle.TotalMileageKm, LastSeen: vehicle.LastSeen, ReportIntervalMs: vehicle.ReportIntervalMs, Status: monitorVehicleStatus(vehicle), }) } if zoom >= monitorPointZoom && len(points) <= monitorPointLimit { result.Mode = "points" result.Points = points return result, nil } type clusterBucket struct { cluster MonitorMapCluster points []MonitorMapPoint } cellSize := 360 * monitorClusterGridPixels / (256 * math.Pow(2, float64(zoom))) if cellSize < 0.01 { cellSize = 0.01 } clusters := map[string]*clusterBucket{} for { clusters = make(map[string]*clusterBucket) for _, point := range points { x := int(math.Floor(point.Longitude / cellSize)) y := int(math.Floor(point.Latitude / cellSize)) key := strconv.Itoa(x) + ":" + strconv.Itoa(y) bucket := clusters[key] if bucket == nil { bucket = &clusterBucket{cluster: MonitorMapCluster{ID: key}} clusters[key] = bucket } cluster := &bucket.cluster bucket.points = append(bucket.points, point) cluster.Count++ cluster.Longitude += (point.Longitude - cluster.Longitude) / float64(cluster.Count) cluster.Latitude += (point.Latitude - cluster.Latitude) / float64(cluster.Count) switch point.Status { case "driving": cluster.Driving++ cluster.Online++ case "idle": cluster.Idle++ cluster.Online++ case "offline": cluster.Offline++ default: cluster.Unknown++ } } if len(clusters) <= monitorPointLimit || cellSize >= 180 { break } cellSize *= 2 } keys := make([]string, 0, len(clusters)) for key := range clusters { keys = append(keys, key) } sort.Strings(keys) for _, key := range keys { bucket := clusters[key] if bucket.cluster.Count == 1 { result.Points = append(result.Points, bucket.points[0]) continue } result.Clusters = append(result.Clusters, bucket.cluster) } if len(result.Points) > 0 { if len(result.Clusters) > 0 { result.Mode = "mixed" } else { result.Mode = "points" } } return result, nil } func (s *Service) Vehicles(ctx context.Context, query url.Values) (Page[VehicleRow], error) { return s.store.Vehicles(ctx, query) } func (s *Service) ResolveVehicleIdentity(ctx context.Context, keyword string, protocol string) (VehicleIdentityResolution, error) { keyword = strings.TrimSpace(keyword) if keyword == "" { return VehicleIdentityResolution{LookupKey: keyword, Protocols: []string{}}, nil } vehicleQuery := url.Values{"keyword": {keyword}, "limit": {"20"}} if protocol = strings.TrimSpace(protocol); protocol != "" { vehicleQuery.Set("protocol", protocol) } vehicles, err := s.store.Vehicles(ctx, vehicleQuery) if err != nil { return VehicleIdentityResolution{}, err } return buildVehicleIdentityResolution(keyword, vehicles.Items), nil } func (s *Service) VehicleCoverage(ctx context.Context, query url.Values) (Page[VehicleCoverageRow], error) { return s.store.VehicleCoverage(ctx, query) } func (s *Service) VehicleCoverageSummary(ctx context.Context, query url.Values) (VehicleCoverageSummary, error) { return s.store.VehicleCoverageSummary(ctx, query) } func (s *Service) VehicleServiceSummary(ctx context.Context) (VehicleServiceSummary, error) { return s.store.VehicleServiceSummary(ctx) } func (s *Service) SourceReadiness(ctx context.Context) (SourceReadinessPlan, error) { summary, err := s.store.VehicleServiceSummary(ctx) if err != nil { return SourceReadinessPlan{}, err } health, err := s.store.OpsHealth(ctx) if err != nil { return SourceReadinessPlan{}, err } quality, err := s.store.QualitySummary(ctx, url.Values{}) if err != nil { return SourceReadinessPlan{}, err } health.Runtime = s.runtime return buildSourceReadinessPlan(summary, health, quality), nil } func (s *Service) VehicleRealtime(ctx context.Context, query url.Values) (Page[VehicleRealtimeRow], error) { resolvedQuery, err := s.resolveVehicleQuery(ctx, query) if err != nil { return Page[VehicleRealtimeRow]{}, err } return s.store.VehicleRealtime(ctx, resolvedQuery) } func (s *Service) VehicleServiceOverview(ctx context.Context, keyword string, protocol string) (VehicleServiceOverview, error) { keyword = strings.TrimSpace(keyword) protocol = strings.TrimSpace(protocol) if batchStore, ok := s.store.(VehicleOverviewBatchStore); ok { page, err := batchStore.VehicleServiceOverviews(ctx, VehicleOverviewBatchQuery{ Keywords: []string{keyword}, Protocol: protocol, Limit: 1, }) if err != nil { return VehicleServiceOverview{}, err } if len(page.Items) > 0 { return page.Items[0], nil } return *buildVehicleServiceOverview("", keyword, &VehicleIdentityResolution{LookupKey: keyword, Protocols: []string{}}, nil, nil, nil, Page[HistoryLocationRow]{}, Page[RawFrameRow]{}, Page[DailyMileageRow]{}, Page[QualityIssueRow]{}), nil } vehicleQuery := url.Values{"keyword": {keyword}, "limit": {"20"}} if protocol != "" { vehicleQuery.Set("protocol", protocol) } vehicles, err := s.store.Vehicles(ctx, vehicleQuery) if err != nil { return VehicleServiceOverview{}, err } resolution := buildVehicleIdentityResolution(keyword, vehicles.Items) identity := resolveVehicleIdentity(keyword, vehicles.Items) resolvedVIN := "" if identity != nil && strings.TrimSpace(identity.VIN) != "" { resolvedVIN = identity.VIN } else if resolution.Resolved { resolvedVIN = resolution.VIN } if resolvedVIN == "" { return *buildVehicleServiceOverview("", keyword, &resolution, nil, nil, nil, Page[HistoryLocationRow]{}, Page[RawFrameRow]{}, Page[DailyMileageRow]{}, Page[QualityIssueRow]{}), nil } realtimeQuery := url.Values{"vin": {resolvedVIN}, "limit": {"1"}} if protocol != "" { realtimeQuery.Set("protocol", protocol) } realtimeSummary, err := s.store.VehicleRealtime(ctx, realtimeQuery) if err != nil { return VehicleServiceOverview{}, err } var summary *VehicleRealtimeRow if len(realtimeSummary.Items) > 0 { summary = &realtimeSummary.Items[0] } sourceStatus := vehicleSourceStatus(vehicles.Items, nil, nil, nil, nil) overview := buildVehicleServiceOverview(resolvedVIN, keyword, &resolution, identity, summary, sourceStatus, Page[HistoryLocationRow]{}, Page[RawFrameRow]{}, Page[DailyMileageRow]{}, Page[QualityIssueRow]{}) return *overview, nil } func (s *Service) VehicleServiceOverviews(ctx context.Context, query VehicleOverviewBatchQuery) (Page[VehicleServiceOverview], error) { keywords, total, limit, offset := overviewBatchPage(query) if offset >= total { return Page[VehicleServiceOverview]{Items: []VehicleServiceOverview{}, Total: total, Limit: limit, Offset: offset}, nil } query.Keywords = keywords query.Limit = limit query.Offset = offset if batchStore, ok := s.store.(VehicleOverviewBatchStore); ok { page, err := batchStore.VehicleServiceOverviews(ctx, query) if err != nil { return Page[VehicleServiceOverview]{}, err } page.Total = total page.Limit = limit page.Offset = offset return page, nil } items := make([]VehicleServiceOverview, 0, len(keywords)) for _, keyword := range keywords { overview, err := s.VehicleServiceOverview(ctx, keyword, query.Protocol) if err != nil { return Page[VehicleServiceOverview]{}, err } items = append(items, overview) } return Page[VehicleServiceOverview]{Items: items, Total: total, Limit: limit, Offset: offset}, nil } func (s *Service) VehicleDetail(ctx context.Context, vin string, protocol string) (VehicleDetail, error) { keyword := strings.TrimSpace(vin) protocol = strings.TrimSpace(protocol) vehicleQuery := url.Values{"keyword": {keyword}, "limit": {"10"}} vehicles, err := s.store.Vehicles(ctx, vehicleQuery) if err != nil { return VehicleDetail{}, err } resolution := buildVehicleIdentityResolution(keyword, vehicles.Items) identity := resolveVehicleIdentity(keyword, vehicles.Items) resolvedVIN := "" if identity != nil && strings.TrimSpace(identity.VIN) != "" { resolvedVIN = identity.VIN } else if resolution.Resolved { resolvedVIN = resolution.VIN } queryVIN := resolvedVIN if queryVIN == "" { queryVIN = keyword } if resolvedVIN == "" { qualityQuery := url.Values{"keyword": {keyword}, "limit": {"20"}} if protocol != "" { qualityQuery.Set("protocol", protocol) } quality, err := s.store.QualityIssues(ctx, qualityQuery) if err != nil { return VehicleDetail{}, err } return VehicleDetail{ VIN: "", LookupKey: keyword, LookupResolved: false, Resolution: &resolution, ServiceStatus: buildVehicleServiceStatus(false, nil), ServiceOverview: buildVehicleServiceOverview("", keyword, &resolution, nil, nil, nil, Page[HistoryLocationRow]{}, Page[RawFrameRow]{}, Page[DailyMileageRow]{}, quality), Sources: []string{}, SourceStatus: []VehicleSourceStatus{}, Realtime: []RealtimeLocationRow{}, History: Page[HistoryLocationRow]{Items: []HistoryLocationRow{}, Limit: 20, Offset: 0}, Raw: Page[RawFrameRow]{Items: []RawFrameRow{}, Limit: 10, Offset: 0}, Mileage: Page[DailyMileageRow]{Items: []DailyMileageRow{}, Limit: 20, Offset: 0}, Quality: quality, }, nil } realtimeQuery := url.Values{"vin": {queryVIN}, "limit": {"20"}} historyQuery := url.Values{"vin": {queryVIN}, "limit": {"20"}} rawQuery := RawFrameQuery{VIN: queryVIN, IncludeFields: true, Limit: 10} mileageQuery := url.Values{"vin": {queryVIN}, "limit": {"20"}} qualityQuery := url.Values{"vin": {queryVIN}, "limit": {"20"}} if protocol != "" { realtimeQuery.Set("protocol", protocol) historyQuery.Set("protocol", protocol) rawQuery.Protocol = protocol qualityQuery.Set("protocol", protocol) } realtimeSummary, err := s.store.VehicleRealtime(ctx, url.Values{"vin": {queryVIN}, "limit": {"1"}}) if err != nil { return VehicleDetail{}, err } var summary *VehicleRealtimeRow if len(realtimeSummary.Items) > 0 { summary = &realtimeSummary.Items[0] } realtime, err := s.store.RealtimeLocations(ctx, realtimeQuery) if err != nil { return VehicleDetail{}, err } history, err := s.store.HistoryLocationsFromTDengine(ctx, historyQuery) if err != nil { return VehicleDetail{}, err } raw, err := s.RawFrames(ctx, rawQuery) if err != nil { return VehicleDetail{}, err } mileage, err := s.store.DailyMileage(ctx, mileageQuery) if err != nil { return VehicleDetail{}, err } quality, err := s.store.QualityIssues(ctx, qualityQuery) if err != nil { return VehicleDetail{}, err } sourceStatus := vehicleSourceStatus(vehicles.Items, realtime.Items, history.Items, raw.Items, mileage.Items) sourceStatus = s.enrichVehicleSourceStatus(ctx, queryVIN, sourceStatus) sourceStatus = completeVehicleSourceStatus(sourceStatus, protocol) serviceStatus := buildVehicleServiceStatus(true, sourceStatus) resolution.ServiceStatus = serviceStatus if identity != nil { identity.ServiceStatus = serviceStatus } if summary != nil { summary.ServiceStatus = serviceStatus } serviceOverview := buildVehicleServiceOverview(resolvedVIN, keyword, &resolution, identity, summary, sourceStatus, history, raw, mileage, quality) serviceOverview.ServiceStatus = serviceStatus profile, err := s.VehicleProfile(ctx, resolvedVIN) if err != nil { return VehicleDetail{}, err } return VehicleDetail{ VIN: resolvedVIN, LookupKey: keyword, LookupResolved: resolvedVIN != "", Resolution: &resolution, Identity: identity, Profile: &profile, RealtimeSummary: summary, ServiceStatus: serviceStatus, ServiceOverview: serviceOverview, SourceConsistency: buildVehicleSourceConsistency(sourceStatus, realtime.Items, protocol), Sources: sourceNames(sourceStatus), SourceStatus: sourceStatus, Realtime: realtime.Items, History: history, Raw: raw, Mileage: mileage, Quality: quality, }, nil } func buildVehicleSourceConsistency(statuses []VehicleSourceStatus, realtime []RealtimeLocationRow, protocol string) *VehicleSourceConsistency { consistency := &VehicleSourceConsistency{ SourceCount: len(statuses), Scope: "all_sources", } if strings.TrimSpace(protocol) != "" { consistency.Scope = "single_source" } if consistency.SourceCount == 0 { seen := map[string]struct{}{} for _, row := range realtime { if strings.TrimSpace(row.Protocol) == "" { continue } seen[row.Protocol] = struct{}{} } consistency.SourceCount = len(seen) } for _, status := range statuses { if status.Online { consistency.OnlineSourceCount++ } if vehicleSourceEvidenceMissing(status) { consistency.MissingProtocols = append(consistency.MissingProtocols, status.Protocol) } } var minMileage, maxMileage float64 hasMileage := false var minTime, maxTime time.Time hasTime := false for _, row := range realtime { if row.Longitude != 0 || row.Latitude != 0 { consistency.LocatedSourceCount++ } if row.TotalMileageKm > 0 { if !hasMileage { minMileage = row.TotalMileageKm maxMileage = row.TotalMileageKm hasMileage = true } else { minMileage = math.Min(minMileage, row.TotalMileageKm) maxMileage = math.Max(maxMileage, row.TotalMileageKm) } } if parsed, ok := parseVehicleServiceTime(row.LastSeen); ok { if !hasTime { minTime = parsed maxTime = parsed hasTime = true } else { if parsed.Before(minTime) { minTime = parsed } if parsed.After(maxTime) { maxTime = parsed } } } } for _, status := range statuses { if parsed, ok := parseVehicleServiceTime(status.LastSeen); ok { if !hasTime { minTime = parsed maxTime = parsed hasTime = true } else { if parsed.Before(minTime) { minTime = parsed } if parsed.After(maxTime) { maxTime = parsed } } } } if hasMileage { consistency.MileageDeltaKm = maxMileage - minMileage } if hasTime { consistency.SourceTimeDeltaSeconds = maxTime.Sub(minTime).Seconds() } applyVehicleSourceConsistencyDiagnosis(consistency) return consistency } func vehicleSourceEvidenceMissing(status VehicleSourceStatus) bool { return strings.TrimSpace(status.Protocol) != "" && !status.Online && strings.TrimSpace(status.LastSeen) == "" && !status.HasRealtime && !status.HasHistory && !status.HasRaw && !status.HasMileage } func applyVehicleSourceConsistencyDiagnosis(consistency *VehicleSourceConsistency) { if consistency == nil { return } switch { case consistency.SourceCount <= 0: consistency.Status = "no_source" consistency.Severity = "warning" consistency.Title = "暂无来源" consistency.Detail = "当前车辆没有可用于交叉校验的数据来源。" case consistency.SourceCount == 1: consistency.Status = "single_source" consistency.Severity = "warning" consistency.Title = "单一来源" consistency.Detail = "当前车辆只有一个数据来源,无法做跨来源一致性校验。" case consistency.OnlineSourceCount <= 0: consistency.Status = "offline" consistency.Severity = "error" consistency.Title = "全部来源离线" consistency.Detail = sourceCoverageDetail(consistency.SourceCount, consistency.OnlineSourceCount, "无法判断实时一致性。") case len(consistency.MissingProtocols) > 0: consistency.Status = "degraded" consistency.Severity = "warning" consistency.Title = "来源不完整" consistency.Detail = sourceCoverageDetail(consistency.SourceCount, consistency.OnlineSourceCount, "车辆服务可用但缺少 "+strings.Join(consistency.MissingProtocols, "、")+" 来源。") case consistency.OnlineSourceCount < consistency.SourceCount: consistency.Status = "degraded" consistency.Severity = "warning" consistency.Title = "来源不完整" consistency.Detail = sourceCoverageDetail(consistency.SourceCount, consistency.OnlineSourceCount, "车辆服务可用但需要关注离线来源。") case consistency.MileageDeltaKm > 5: consistency.Status = "mileage_divergent" consistency.Severity = "warning" consistency.Title = "里程差异偏大" consistency.Detail = "多个来源的实时总里程差异超过 5 km,请核对来源解析、单位和上报时间。" case consistency.SourceTimeDeltaSeconds > 300: consistency.Status = "time_divergent" consistency.Severity = "warning" consistency.Title = "来源时间差偏大" consistency.Detail = "多个来源的最新上报时间相差超过 5 分钟,请关注转发延迟或链路断点。" default: consistency.Status = "consistent" consistency.Severity = "ok" consistency.Title = "来源一致" consistency.Detail = sourceCoverageDetail(consistency.SourceCount, consistency.OnlineSourceCount, "多源实时状态一致。") } } func parseVehicleServiceTime(value string) (time.Time, bool) { value = strings.TrimSpace(value) if value == "" { return time.Time{}, false } layouts := []string{time.RFC3339, "2006-01-02 15:04:05", "2006-01-02T15:04:05-07:00"} for _, layout := range layouts { if parsed, err := time.Parse(layout, value); err == nil { return parsed, true } } return time.Time{}, false } func buildVehicleServiceOverview(vin string, lookupKey string, resolution *VehicleIdentityResolution, identity *VehicleRow, summary *VehicleRealtimeRow, statuses []VehicleSourceStatus, history Page[HistoryLocationRow], raw Page[RawFrameRow], mileage Page[DailyMileageRow], quality Page[QualityIssueRow]) *VehicleServiceOverview { overview := &VehicleServiceOverview{ VIN: strings.TrimSpace(vin), Plate: "", Phone: "", OEM: "", Protocols: sourceNames(statuses), CoverageStatus: "no_data", SourceCount: len(statuses), OnlineSourceCount: 0, HistoryCount: pageCount(history.Total, len(history.Items)), RawCount: pageCount(raw.Total, len(raw.Items)), MileageCount: pageCount(mileage.Total, len(mileage.Items)), QualityIssueCount: pageCount(quality.Total, len(quality.Items)), } if overview.VIN == "" && resolution != nil && resolution.Resolved { overview.VIN = strings.TrimSpace(resolution.VIN) } if overview.VIN == "" { overview.VIN = strings.TrimSpace(lookupKey) } if resolution != nil { overview.Plate = resolution.Plate overview.Phone = resolution.Phone overview.OEM = resolution.OEM if len(overview.Protocols) == 0 { overview.Protocols = append([]string(nil), resolution.Protocols...) overview.SourceCount = len(overview.Protocols) } overview.LastSeen = latestString(overview.LastSeen, resolution.LastSeen) } if identity != nil { overview.Plate = firstNonEmpty(overview.Plate, identity.Plate) overview.Phone = firstNonEmpty(overview.Phone, identity.Phone) overview.OEM = firstNonEmpty(overview.OEM, identity.OEM) overview.LastSeen = latestString(overview.LastSeen, identity.LastSeen) } if summary != nil { overview.Plate = firstNonEmpty(overview.Plate, summary.Plate) overview.Phone = firstNonEmpty(overview.Phone, summary.Phone) overview.OEM = firstNonEmpty(overview.OEM, summary.OEM) overview.PrimaryProtocol = summary.PrimaryProtocol overview.LastSeen = latestString(overview.LastSeen, summary.LastSeen) overview.RealtimeCount = pageCount(summary.SourceCount, len(summary.Protocols)) if len(statuses) == 0 { overview.SourceCount = summary.SourceCount overview.OnlineSourceCount = summary.OnlineSourceCount } if len(overview.Protocols) == 0 { overview.Protocols = append([]string(nil), summary.Protocols...) } } for _, status := range statuses { if status.Online { overview.OnlineSourceCount++ } if overview.PrimaryProtocol == "" && status.Online { overview.PrimaryProtocol = status.Protocol } overview.LastSeen = latestString(overview.LastSeen, status.LastSeen) } if overview.PrimaryProtocol == "" && len(overview.Protocols) > 0 { overview.PrimaryProtocol = overview.Protocols[0] } overview.CoverageStatus = coverageStatus(overview.SourceCount, overview.OnlineSourceCount) resolved := identity != nil && strings.TrimSpace(identity.VIN) != "" if !resolved && resolution != nil { resolved = resolution.Resolved } applyVehicleServiceOverviewDerivedFields(overview, resolved) overview.SourceConsistency.MissingProtocols = missingProtocolsFromStatuses(statuses) applyVehicleSourceConsistencyDiagnosis(overview.SourceConsistency) return overview } func applyVehicleServiceOverviewDerivedFields(overview *VehicleServiceOverview, resolved bool) { if overview == nil { return } overview.CoverageStatus = coverageStatus(overview.SourceCount, overview.OnlineSourceCount) overview.ServiceStatus = buildVehicleServiceStatus(resolved, statusesForOverview(overview)) overview.SourceConsistency = &VehicleSourceConsistency{ SourceCount: overview.SourceCount, OnlineSourceCount: overview.OnlineSourceCount, MissingProtocols: missingCanonicalProtocols(overview.Protocols), Scope: "all_sources", } applyVehicleSourceConsistencyDiagnosis(overview.SourceConsistency) } func statusesForOverview(overview *VehicleServiceOverview) []VehicleSourceStatus { statuses := make([]VehicleSourceStatus, 0, overview.SourceCount) for index, protocol := range overview.Protocols { statuses = append(statuses, VehicleSourceStatus{ Protocol: protocol, Online: index < overview.OnlineSourceCount, LastSeen: overview.LastSeen, }) } for len(statuses) < overview.SourceCount { statuses = append(statuses, VehicleSourceStatus{ Protocol: "UNKNOWN", Online: len(statuses) < overview.OnlineSourceCount, LastSeen: overview.LastSeen, }) } return statuses } func pageCount(total int, itemCount int) int { if total > 0 { return total } return itemCount } func coverageStatus(sourceCount int, onlineSourceCount int) string { if sourceCount <= 0 { return "no_data" } if onlineSourceCount <= 0 { return "offline" } if onlineSourceCount < sourceCount { return "partial" } return "online" } func (s *Service) enrichVehicleSourceStatus(ctx context.Context, vin string, statuses []VehicleSourceStatus) []VehicleSourceStatus { for index := range statuses { protocol := statuses[index].Protocol if protocol == "" { continue } if !statuses[index].HasHistory { query := url.Values{"vin": {vin}, "protocol": {protocol}, "limit": {"1"}} if page, err := s.store.HistoryLocationsFromTDengine(ctx, query); err == nil && len(page.Items) > 0 { statuses[index].HasHistory = true statuses[index].LastSeen = latestString(statuses[index].LastSeen, firstNonEmpty(page.Items[0].ServerTime, page.Items[0].DeviceTime)) } } if !statuses[index].HasRaw { page, err := s.store.RawFrames(ctx, RawFrameQuery{VIN: vin, Protocol: protocol, Limit: 1}) if err == nil && len(page.Items) > 0 { statuses[index].HasRaw = true statuses[index].LastSeen = latestString(statuses[index].LastSeen, firstNonEmpty(page.Items[0].ServerTime, page.Items[0].DeviceTime)) } } } return statuses } func resolveVehicleIdentity(keyword string, vehicles []VehicleRow) *VehicleRow { keyword = strings.TrimSpace(keyword) if keyword == "" { return nil } for index := range vehicles { if strings.EqualFold(vehicles[index].VIN, keyword) { return &vehicles[index] } } for index := range vehicles { if strings.EqualFold(vehicles[index].Plate, keyword) || strings.EqualFold(vehicles[index].Phone, keyword) { return &vehicles[index] } } if len(vehicles) > 0 { return &vehicles[0] } return nil } func isLikelyVIN(value string) bool { value = strings.ToUpper(strings.TrimSpace(value)) if len(value) != 17 { return false } for _, char := range value { if char >= '0' && char <= '9' { continue } if char >= 'A' && char <= 'Z' && char != 'I' && char != 'O' && char != 'Q' { continue } return false } return true } func (s *Service) RealtimeLocations(ctx context.Context, query url.Values) (Page[RealtimeLocationRow], error) { resolvedQuery, err := s.resolveVehicleQuery(ctx, query) if err != nil { return Page[RealtimeLocationRow]{}, err } return s.store.RealtimeLocations(ctx, resolvedQuery) } func (s *Service) HistoryLocations(ctx context.Context, query url.Values) (Page[HistoryLocationRow], error) { resolvedQuery, err := s.resolveVehicleQuery(ctx, query) if err != nil { return Page[HistoryLocationRow]{}, err } return s.store.HistoryLocationsFromTDengine(ctx, resolvedQuery) } func (s *Service) TrackPlayback(ctx context.Context, query url.Values) (TrackPlaybackResponse, error) { keyword := strings.TrimSpace(firstNonEmpty(query.Get("keyword"), query.Get("vin"))) if keyword == "" { return TrackPlaybackResponse{}, clientError{Code: "VEHICLE_KEY_REQUIRED", Message: "轨迹查询需要车牌、VIN 或终端标识"} } if err := validateTrackWindow(query.Get("dateFrom"), query.Get("dateTo")); err != nil { return TrackPlaybackResponse{}, err } maxPoints := parsePositive(query.Get("maxPoints"), 1200) if maxPoints < 2 { maxPoints = 2 } if maxPoints > 2000 { maxPoints = 2000 } resolvedQuery, err := s.resolveVehicleQuery(ctx, query) if err != nil { return TrackPlaybackResponse{}, err } resolvedQuery.Set("limit", "5000") resolvedQuery.Set("offset", "0") page, err := s.store.HistoryLocationsFromTDengine(ctx, resolvedQuery) if err != nil { return TrackPlaybackResponse{}, err } rawPoints := append([]HistoryLocationRow(nil), page.Items...) sort.SliceStable(rawPoints, func(i, j int) bool { left, leftOK := parseVehicleServiceTime(rawPoints[i].DeviceTime) right, rightOK := parseVehicleServiceTime(rawPoints[j].DeviceTime) if leftOK && rightOK && !left.Equal(right) { return left.Before(right) } return rawPoints[i].DeviceTime < rawPoints[j].DeviceTime }) cleanPoints, quality := analyzeTrackPoints(rawPoints) points, selectedProtocol, alternateSourcePoints := selectTrackPrimarySource(cleanPoints, query.Get("protocol")) quality.SelectedProtocol = selectedProtocol quality.AlternateSourcePoints = alternateSourcePoints quality.ValidPoints = len(points) applyTrackSequenceQuality(points, &quality) result := TrackPlaybackResponse{ Points: []HistoryLocationRow{}, Events: []TrackPlaybackEvent{}, Sources: summarizeTrackSources(cleanPoints), Segments: []TrackSegment{}, Stops: []TrackStop{}, Total: page.Total, Truncated: page.Total > len(rawPoints), Quality: quality, AsOf: time.Now().UTC().Format(time.RFC3339), } result.Coverage = trackCoverage(query, page.Total, len(rawPoints), len(points), 0, result.Truncated, false) applyTrackPrimarySourceCoverage(&result.Coverage, selectedProtocol, alternateSourcePoints) if len(points) == 0 { return result, nil } result.VIN = points[0].VIN result.Plate = points[0].Plate segments := buildTrackSegments(points) stops := buildTrackStops(points, segments) result.Summary = summarizeTrack(points) result.Coverage.ActualStart = result.Summary.StartTime result.Coverage.ActualEnd = result.Summary.EndTime result.Summary.SegmentCount = len(segments) result.Summary.StopCount = len(stops) for _, segment := range segments { switch segment.Type { case "moving": result.Summary.MovingSeconds += segment.DurationSeconds case "stopped": result.Summary.StoppedSeconds += segment.DurationSeconds } } result.Segments = segments result.Stops = stops result.Events = buildTrackEvents(points, segments, stops) keyIndexes := trackKeyIndexes(result.Events, segments, stops) var sampledIndexes []int result.Points, sampledIndexes = sampleTrackPointsWithKeys(points, maxPoints, keyIndexes) result.Sampled = len(result.Points) < len(points) for index := range result.Events { result.Events[index].SampledIndex = nearestSampledIndex(result.Events[index].Index, sampledIndexes) } for index := range result.Segments { result.Segments[index].SampledStartIndex = nearestSampledIndex(result.Segments[index].StartIndex, sampledIndexes) result.Segments[index].SampledEndIndex = nearestSampledIndex(result.Segments[index].EndIndex, sampledIndexes) } for index := range result.Stops { result.Stops[index].SampledIndex = nearestSampledIndex((result.Stops[index].startIndex+result.Stops[index].endIndex)/2, sampledIndexes) } result.Coverage = trackCoverage(query, page.Total, len(rawPoints), len(points), len(result.Points), result.Truncated, result.Sampled) applyTrackPrimarySourceCoverage(&result.Coverage, selectedProtocol, alternateSourcePoints) result.Coverage.ActualStart = result.Summary.StartTime result.Coverage.ActualEnd = result.Summary.EndTime return result, nil } func validateTrackWindow(dateFrom, dateTo string) error { dateFrom = strings.TrimSpace(dateFrom) dateTo = strings.TrimSpace(dateTo) if dateFrom == "" && dateTo == "" { return nil } if dateFrom == "" || dateTo == "" { return clientError{Code: "TRACK_TIME_RANGE_REQUIRED", Message: "轨迹时间范围需要同时提供开始和结束时间"} } start, startOK := parseTrackRequestTime(dateFrom) end, endOK := parseTrackRequestTime(dateTo) if !startOK || !endOK { return clientError{Code: "TRACK_TIME_INVALID", Message: "轨迹时间格式无效"} } if !end.After(start) { return clientError{Code: "TRACK_TIME_ORDER_INVALID", Message: "轨迹结束时间必须晚于开始时间"} } if end.Sub(start) > 7*24*time.Hour { return clientError{Code: "TRACK_TIME_RANGE_TOO_LARGE", Message: "单次轨迹回放最长支持 7 天,请缩小时间范围"} } return nil } func parseTrackRequestTime(value string) (time.Time, bool) { shanghai := time.FixedZone("Asia/Shanghai", 8*60*60) for _, layout := range []string{time.RFC3339, "2006-01-02T15:04", "2006-01-02T15:04:05", "2006-01-02 15:04:05"} { if layout == time.RFC3339 { if parsed, err := time.Parse(layout, value); err == nil { return parsed, true } continue } if parsed, err := time.ParseInLocation(layout, value, shanghai); err == nil { return parsed, true } } return time.Time{}, false } func (s *Service) HistoryMetricCatalog() HistoryMetricCatalog { return HistoryMetricCatalog{ Categories: []HistoryDataCategory{{Key: "location", Label: "位置数据"}, {Key: "raw", Label: "原始报文"}, {Key: "mileage", Label: "日里程"}}, Metrics: append(append(append([]HistoryMetricDefinition{}, historyLocationMetrics()...), historyRawMetrics()...), historyMileageMetrics()...), } } type metricCatalogStore interface { MetricDefinitions(context.Context) ([]MetricDefinition, error) } func (s *Service) metricDefinitions(ctx context.Context) ([]MetricDefinition, error) { if store, ok := s.store.(metricCatalogStore); ok { return store.MetricDefinitions(ctx) } return metricDefinitions(), nil } func (s *Service) MetricCatalog(ctx context.Context) (MetricCatalog, error) { definitions, err := s.metricDefinitions(ctx) if err != nil { return MetricCatalog{}, err } return MetricCatalog{Metrics: definitions, AsOf: time.Now().UTC().Format(time.RFC3339)}, nil } const latestTelemetryStaleAfter = 5 * time.Minute type latestTelemetryRawResult struct { page Page[RawFrameRow] err error } type latestTelemetryCatalogResult struct { definitions []MetricDefinition err error } // LatestTelemetry keeps the latest scalar value for every unified metric or // source field. RAW and catalog reads are independent, so they run in parallel. func (s *Service) LatestTelemetry(ctx context.Context, vehicleKey string) (LatestTelemetryResponse, error) { vehicleKey = strings.TrimSpace(vehicleKey) if vehicleKey == "" { return LatestTelemetryResponse{}, clientError{Code: "VEHICLE_KEY_REQUIRED", Message: "最新遥测查询需要车牌、VIN 或终端标识"} } resolvedVIN, err := s.resolveVehicleVIN(ctx, vehicleKey, "") if err != nil { return LatestTelemetryResponse{}, err } rawResult := make(chan latestTelemetryRawResult, len(canonicalVehicleProtocols)) catalogResult := make(chan latestTelemetryCatalogResult, 1) for _, protocol := range canonicalVehicleProtocols { go func(protocol string) { page, queryErr := s.store.RawFrames(ctx, RawFrameQuery{VIN: resolvedVIN, Protocol: protocol, IncludeFields: true, Limit: 5, SkipCount: true}) rawResult <- latestTelemetryRawResult{page: page, err: queryErr} }(protocol) } go func() { definitions, err := s.metricDefinitions(ctx) catalogResult <- latestTelemetryCatalogResult{definitions: definitions, err: err} }() frames := make([]RawFrameRow, 0, len(canonicalVehicleProtocols)*5) for range canonicalVehicleProtocols { raw := <-rawResult if raw.err != nil { return LatestTelemetryResponse{}, raw.err } frames = append(frames, raw.page.Items...) } catalog := <-catalogResult if catalog.err != nil { return LatestTelemetryResponse{}, catalog.err } return buildLatestTelemetryResponse(resolvedVIN, frames, catalog.definitions, time.Now()), nil } type latestTelemetryCandidate struct { value LatestTelemetryValue sortTime time.Time timeValid bool } func buildLatestTelemetryResponse(vehicleKey string, frames []RawFrameRow, definitions []MetricDefinition, now time.Time) LatestTelemetryResponse { asOf := now.UTC().Format(time.RFC3339) response := LatestTelemetryResponse{ VIN: strings.TrimSpace(vehicleKey), Categories: []LatestTelemetryCategory{}, Values: []LatestTelemetryValue{}, AsOf: asOf, StaleAfterSeconds: int64(latestTelemetryStaleAfter / time.Second), ScannedFrames: len(frames), } definitionBySource := make(map[string]MetricDefinition, len(definitions)*2) for _, definition := range definitions { for protocol, sourceField := range definition.SourceFields { definitionBySource[telemetrySourceLookupKey(protocol, sourceField)] = definition } } candidates := make(map[string]latestTelemetryCandidate, 64) for _, frame := range frames { if response.VIN == vehicleKey && strings.TrimSpace(frame.VIN) != "" { response.VIN = strings.TrimSpace(frame.VIN) } observed, observedOK := latestTelemetryObservedTime(frame) quality, qualityReason, freshness, delay := latestTelemetryQuality(frame, now) normalizedProtocol := strings.ToUpper(strings.TrimSpace(frame.Protocol)) sourceLookupPrefix := normalizedProtocol + "\x00" for sourceField, rawValue := range frame.ParsedFields { if !isTelemetryScalar(rawValue) { continue } definition, unified := definitionBySource[sourceLookupPrefix+strings.TrimSpace(sourceField)] key := sourceField if unified { key = definition.Key } if previous, exists := candidates[key]; exists && (!observedOK || (previous.timeValid && !observed.After(previous.sortTime))) { continue } label, unit := latestTelemetryRawMetadata(sourceField) category := telemetrySourceCategory(sourceField) valueType := "dynamic" description := "" value := rawValue if unified { label = definition.Label unit = definition.Unit category = normalizeTelemetryCategory(definition.Category) valueType = definition.ValueType description = definition.Description value = normalizeTelemetryValue(rawValue, definition.ValueType) } candidates[key] = latestTelemetryCandidate{ value: LatestTelemetryValue{ Key: key, SourceField: sourceField, Label: label, Description: description, Unit: unit, Category: category, ValueType: valueType, Value: value, Protocol: normalizedProtocol, SourceEndpoint: frame.SourceEndpoint, FrameID: frame.ID, DeviceTime: frame.DeviceTime, ServerTime: frame.ServerTime, Quality: quality, QualityReason: qualityReason, FreshnessSeconds: freshness, DataDelaySeconds: delay, }, sortTime: observed, timeValid: observedOK, } } } for _, candidate := range candidates { response.Values = append(response.Values, candidate.value) } sort.Slice(response.Values, func(i, j int) bool { left, right := telemetryCategoryRank(response.Values[i].Category), telemetryCategoryRank(response.Values[j].Category) if left != right { return left < right } if response.Values[i].Label != response.Values[j].Label { return response.Values[i].Label < response.Values[j].Label } return response.Values[i].Key < response.Values[j].Key }) categoryIndexes := map[string]int{} for _, value := range response.Values { index, exists := categoryIndexes[value.Category] if !exists { categoryIndexes[value.Category] = len(response.Categories) response.Categories = append(response.Categories, LatestTelemetryCategory{Key: value.Category, Label: telemetryCategoryLabel(value.Category)}) index = len(response.Categories) - 1 } response.Categories[index].Count++ } response.Evidence = fmt.Sprintf("scanned %d newest RAW frames; selected %d newest scalar values by unified metric/source field", len(frames), len(response.Values)) return response } func telemetrySourceLookupKey(protocol, sourceField string) string { return strings.ToUpper(strings.TrimSpace(protocol)) + "\x00" + strings.TrimSpace(sourceField) } func latestTelemetryObservedTime(frame RawFrameRow) (time.Time, bool) { if parsed, ok := parseLatestTelemetryTime(frame.ServerTime); ok { return parsed, true } return parseLatestTelemetryTime(frame.DeviceTime) } func parseLatestTelemetryTime(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 } } shanghai := time.FixedZone("Asia/Shanghai", 8*60*60) for _, layout := range []string{"2006-01-02 15:04:05.999999999", "2006-01-02 15:04:05", "2006-01-02T15:04:05.999999999", "2006-01-02T15:04:05"} { if parsed, err := time.ParseInLocation(layout, value, shanghai); err == nil { return parsed, true } } return time.Time{}, false } func latestTelemetryQuality(frame RawFrameRow, now time.Time) (string, string, int64, *int64) { serverTime, serverOK := parseLatestTelemetryTime(frame.ServerTime) deviceTime, deviceOK := parseLatestTelemetryTime(frame.DeviceTime) var delay *int64 if serverOK && deviceOK { seconds := int64(math.Abs(serverTime.Sub(deviceTime).Seconds())) delay = &seconds } if status := strings.TrimSpace(frame.ParseStatus); status != "" && !strings.EqualFold(status, "ok") { reason := "原始帧解析状态为 " + status if detail := strings.TrimSpace(frame.ParseError); detail != "" { reason += ":" + detail } return "warning", reason, telemetryFreshnessSeconds(serverTime, serverOK, now), delay } if !serverOK { return "warning", "缺少可解析的服务接收时间", 0, delay } if serverTime.After(now.Add(latestTelemetryStaleAfter)) { return "warning", "服务接收时间晚于平台当前时间超过 5 分钟", 0, delay } freshness := telemetryFreshnessSeconds(serverTime, true, now) if delay != nil && *delay > int64(latestTelemetryStaleAfter/time.Second) { return "warning", "设备时间与服务接收时间相差超过 5 分钟", freshness, delay } if freshness > int64(latestTelemetryStaleAfter/time.Second) { return "stale", "最新值距平台当前时间超过 5 分钟", freshness, delay } return "good", "解析成功且接收时间在 5 分钟内", freshness, delay } func telemetryFreshnessSeconds(serverTime time.Time, valid bool, now time.Time) int64 { if !valid || serverTime.After(now) { return 0 } return int64(now.Sub(serverTime).Seconds()) } func isTelemetryScalar(value any) bool { switch value.(type) { case nil, map[string]any, []any: return false case string, bool, float64, float32, int, int8, int16, int32, int64, uint, uint8, uint16, uint32, uint64, json.Number: return true default: return false } } func normalizeTelemetryValue(value any, valueType string) any { if valueType != "boolean" { return value } switch typed := value.(type) { case bool: return typed case float64: return typed != 0 case json.Number: number, err := typed.Float64() return err == nil && number != 0 case string: return strings.EqualFold(typed, "true") || typed == "1" default: return value } } func latestTelemetryRawMetadata(sourceField string) (string, string) { label, unit := rawMetricMetadata(sourceField) protocol := strings.ToUpper(strings.Split(sourceField, ".")[0]) label = strings.TrimSuffix(label, " · "+protocol) return label, unit } func normalizeTelemetryCategory(category string) string { switch strings.ToLower(strings.TrimSpace(category)) { case "driving", "vehicle": return "vehicle" case "energy", "battery": return "battery" case "safety", "alarm": return "alarm" case "location": return "location" case "quality": return "quality" default: return "extension" } } func telemetrySourceCategory(sourceField string) string { normalized := strings.ToLower(sourceField) switch { case strings.Contains(normalized, "alarm") || strings.Contains(normalized, "fault") || strings.Contains(normalized, "warning"): return "alarm" case strings.Contains(normalized, "fuel_cell") || strings.Contains(normalized, "fuelcell") || strings.Contains(normalized, "fc_stack"): return "fuel-cell" case strings.Contains(normalized, "motor") || strings.Contains(normalized, "engine") || strings.Contains(normalized, "rpm"): return "motor" case strings.Contains(normalized, "battery") || strings.Contains(normalized, "soc") || strings.Contains(normalized, "cell_") || strings.Contains(normalized, "voltage") || strings.Contains(normalized, "current"): return "battery" case strings.Contains(normalized, "location") || strings.Contains(normalized, "longitude") || strings.Contains(normalized, "latitude") || strings.Contains(normalized, "direction") || strings.Contains(normalized, "altitude"): return "location" case strings.Contains(normalized, ".vehicle.") || strings.HasPrefix(normalized, "vehicle.") || strings.Contains(normalized, "speed") || strings.Contains(normalized, "mileage") || strings.Contains(normalized, "gear") || strings.Contains(normalized, "brake") || strings.Contains(normalized, "accelerator"): return "vehicle" default: return "extension" } } func telemetryCategoryRank(category string) int { for index, key := range []string{"vehicle", "motor", "fuel-cell", "battery", "location", "alarm", "quality", "extension"} { if category == key { return index } } return 99 } func telemetryCategoryLabel(category string) string { labels := map[string]string{"vehicle": "整车数据", "motor": "驱动电机", "fuel-cell": "燃料电池", "battery": "电池", "location": "定位", "alarm": "报警", "quality": "数据质量", "extension": "扩展字段"} if label := labels[category]; label != "" { return label } return category } func metricDefinitions() []MetricDefinition { allProtocols := append([]string(nil), canonicalVehicleProtocols...) return []MetricDefinition{ {Key: "speed_kmh", Label: "速度", Description: "车辆最新行驶速度", Unit: "km/h", Category: "driving", ValueType: "numeric", Protocols: allProtocols, SourceFields: map[string]string{"GB32960": "gb32960.vehicle.speed_kmh", "JT808": "jt808.location.speed_kmh", "YUTONG_MQTT": "yutong_mqtt.data.meter_speed"}, Searchable: true, Chartable: true, Alertable: true}, {Key: "soc_percent", Label: "SOC", Description: "动力电池剩余电量百分比", Unit: "%", Category: "energy", ValueType: "numeric", Protocols: []string{"GB32960", "YUTONG_MQTT"}, SourceFields: map[string]string{"GB32960": "gb32960.vehicle.soc_percent", "YUTONG_MQTT": "yutong_mqtt.data.battery_capacity_soc"}, Searchable: true, Chartable: true, Alertable: true}, {Key: "alarm_active", Label: "协议告警位", Description: "协议报文是否携带活动告警", Unit: "", Category: "safety", ValueType: "boolean", Protocols: []string{"GB32960", "JT808"}, SourceFields: map[string]string{"GB32960": "gb32960.alarm.general_alarm_flag", "JT808": "jt808.location.alarm_flag"}, Searchable: true, Chartable: false, Alertable: true}, {Key: "freshness_sec", Label: "离线时长", Description: "最新数据距当前时间的秒数", Unit: "s", Category: "quality", ValueType: "numeric", Protocols: allProtocols, SourceFields: map[string]string{"PLATFORM": "vehicle_realtime_location.updated_at"}, Searchable: true, Chartable: true, Alertable: true}, {Key: "data_delay_sec", Label: "数据延迟", Description: "服务接收时间与设备事件时间的差值", Unit: "s", Category: "quality", ValueType: "numeric", Protocols: allProtocols, SourceFields: map[string]string{"PLATFORM": "received_at-event_time"}, Searchable: true, Chartable: true, Alertable: true}, {Key: "total_mileage_km", Label: "总里程", Description: "车辆累计行驶里程", Unit: "km", Category: "driving", ValueType: "numeric", Protocols: allProtocols, SourceFields: map[string]string{"GB32960": "gb32960.vehicle.total_mileage_km", "JT808": "jt808.location.total_mileage_km", "YUTONG_MQTT": "yutong.vehicle.total_mileage_km"}, Searchable: true, Chartable: true, Alertable: false}, {Key: "longitude", Label: "经度", Description: "WGS84/协议归一化经度", Unit: "°", Category: "location", ValueType: "numeric", Protocols: allProtocols, SourceFields: map[string]string{"PLATFORM": "vehicle_realtime_location.longitude"}, Searchable: true, Chartable: false, Alertable: false}, {Key: "latitude", Label: "纬度", Description: "WGS84/协议归一化纬度", Unit: "°", Category: "location", ValueType: "numeric", Protocols: allProtocols, SourceFields: map[string]string{"PLATFORM": "vehicle_realtime_location.latitude"}, Searchable: true, Chartable: false, Alertable: false}, } } func (s *Service) HistoryData(ctx context.Context, query url.Values) (HistoryDataResponse, error) { started := time.Now() category := strings.ToLower(strings.TrimSpace(query.Get("category"))) if category == "" { category = "location" } if category != "location" && category != "raw" && category != "mileage" { return HistoryDataResponse{}, clientError{Code: "HISTORY_CATEGORY_INVALID", Message: "历史数据类型仅支持 location、raw 或 mileage"} } keywords := historyQueryKeywords(query) if len(keywords) == 0 { return HistoryDataResponse{}, clientError{Code: "VEHICLE_KEY_REQUIRED", Message: "历史查询至少需要一个车牌、VIN 或终端标识"} } if len(keywords) > 5 { return HistoryDataResponse{}, clientError{Code: "VEHICLE_LIMIT_EXCEEDED", Message: "单次历史查询最多支持 5 台车辆"} } limit := parsePositive(query.Get("limit"), 50) if limit > 200 { limit = 200 } offset := parsePositive(query.Get("offset"), 0) fetchLimit := limit + offset if fetchLimit > 1000 { fetchLimit = 1000 } rows := make([]HistoryDataRow, 0, fetchLimit*len(keywords)) total := 0 for _, keyword := range keywords { vehicleRows, vehicleTotal, err := s.historyDataForVehicle(ctx, category, keyword, query, fetchLimit) if err != nil { return HistoryDataResponse{}, err } rows = append(rows, vehicleRows...) total += vehicleTotal } sort.SliceStable(rows, func(i, j int) bool { if rows[i].DeviceTime == rows[j].DeviceTime { return rows[i].VIN < rows[j].VIN } return rows[i].DeviceTime > rows[j].DeviceTime }) if offset > len(rows) { offset = len(rows) } end := offset + limit if end > len(rows) { end = len(rows) } pageRows := append([]HistoryDataRow(nil), rows[offset:end]...) columns := historyMetricsForCategory(category) if category == "raw" { columns = mergeDiscoveredRawMetrics(columns, pageRows) } vehicles := map[string]struct{}{} sources := map[string]struct{}{} for _, row := range rows { vehicles[row.VIN] = struct{}{} sources[row.Protocol] = struct{}{} } sourceNames := make([]string, 0, len(sources)) for source := range sources { if source != "" { sourceNames = append(sourceNames, source) } } sort.Strings(sourceNames) return HistoryDataResponse{ Category: category, Columns: columns, Rows: pageRows, Total: total, Limit: limit, Offset: offset, Summary: HistoryDataSummary{ResultRows: total, VehicleCount: len(vehicles), Sources: sourceNames, QueryDuration: time.Since(started).Milliseconds()}, AsOf: time.Now().UTC().Format(time.RFC3339), }, nil } type historyLocationSeriesStore interface { HistoryLocationSeries(context.Context, HistoryLocationSeriesQuery) ([]HistoryLocationSeriesBucket, error) } type historySeriesMetricSpec struct { definition HistoryMetricDefinition aggregation string value func(HistoryLocationSeriesBucket) *float64 minimum func(HistoryLocationSeriesBucket) *float64 maximum func(HistoryLocationSeriesBucket) *float64 } func historySeriesMetricSpecs() map[string]historySeriesMetricSpec { return map[string]historySeriesMetricSpec{ "speedKmh": { definition: HistoryMetricDefinition{Key: "speedKmh", Label: "速度", Unit: "km/h", Category: "location", ValueType: "number", DefaultVisible: true}, aggregation: "avg", value: func(row HistoryLocationSeriesBucket) *float64 { return row.SpeedAverage }, minimum: func(row HistoryLocationSeriesBucket) *float64 { return row.SpeedMinimum }, maximum: func(row HistoryLocationSeriesBucket) *float64 { return row.SpeedMaximum }, }, "totalMileageKm": { definition: HistoryMetricDefinition{Key: "totalMileageKm", Label: "总里程", Unit: "km", Category: "location", ValueType: "number", DefaultVisible: true}, aggregation: "last", value: func(row HistoryLocationSeriesBucket) *float64 { return row.MileageLast }, minimum: func(row HistoryLocationSeriesBucket) *float64 { return row.MileageMinimum }, maximum: func(row HistoryLocationSeriesBucket) *float64 { return row.MileageMaximum }, }, } } func (s *Service) HistorySeries(ctx context.Context, query url.Values) (HistorySeriesResponse, error) { started := time.Now() keywords := historyQueryKeywords(query) if len(keywords) == 0 { return HistorySeriesResponse{}, clientError{Code: "VEHICLE_KEY_REQUIRED", Message: "趋势查询至少需要一个车牌、VIN 或终端标识"} } if len(keywords) > 5 { return HistorySeriesResponse{}, clientError{Code: "VEHICLE_LIMIT_EXCEEDED", Message: "单次趋势查询最多支持 5 台车辆"} } category := strings.ToLower(strings.TrimSpace(query.Get("category"))) if category == "" { category = "location" } if category != "location" { return HistorySeriesResponse{}, clientError{Code: "HISTORY_SERIES_CATEGORY_UNSUPPORTED", Message: "当前仅位置数据支持聚合趋势,原始报文和日里程请查看明细"} } dateFrom, dateTo, start, end, err := historySeriesWindow(query.Get("dateFrom"), query.Get("dateTo"), time.Now()) if err != nil { return HistorySeriesResponse{}, err } targetPoints := parsePositive(query.Get("targetPoints"), 240) if targetPoints < 60 { targetPoints = 60 } if targetPoints > 600 { targetPoints = 600 } grainSeconds := historySeriesGrain(end.Sub(start), targetPoints) specs := historySeriesMetricSpecs() metricKeys := historySeriesMetrics(query.Get("metrics"), specs) definitions := make([]HistoryMetricDefinition, 0, len(metricKeys)) for _, key := range metricKeys { definitions = append(definitions, specs[key].definition) } store, ok := s.store.(historyLocationSeriesStore) if !ok { return HistorySeriesResponse{}, fmt.Errorf("history series aggregation is unavailable") } protocol := strings.TrimSpace(query.Get("protocol")) type vehicleBuckets struct { vin, plate string buckets []HistoryLocationSeriesBucket } vehicles := make([]vehicleBuckets, 0, len(keywords)) for _, keyword := range keywords { vin, resolveErr := s.resolveVehicleVIN(ctx, keyword, protocol) if resolveErr != nil { return HistorySeriesResponse{}, resolveErr } buckets, aggregateErr := store.HistoryLocationSeries(ctx, HistoryLocationSeriesQuery{VIN: vin, Protocol: protocol, DateFrom: dateFrom, DateTo: dateTo, GrainSeconds: grainSeconds}) if aggregateErr != nil { return HistorySeriesResponse{}, aggregateErr } vehicles = append(vehicles, vehicleBuckets{vin: vin, plate: s.historySeriesPlate(ctx, vin), buckets: buckets}) } seriesByKey := map[string]*HistorySeries{} seriesOrder := make([]string, 0) groupBuckets := map[string]int{} var rawPointCount int64 for _, vehicle := range vehicles { for _, bucket := range vehicle.buckets { groupKey := vehicle.vin + "\x00" + bucket.Protocol groupBuckets[groupKey]++ rawPointCount += bucket.Count for _, metricKey := range metricKeys { spec := specs[metricKey] key := groupKey + "\x00" + metricKey series := seriesByKey[key] if series == nil { series = &HistorySeries{VIN: vehicle.vin, Plate: vehicle.plate, Protocol: bucket.Protocol, Metric: metricKey, Label: spec.definition.Label, Unit: spec.definition.Unit, Aggregation: spec.aggregation, Points: []HistorySeriesPoint{}} seriesByKey[key] = series seriesOrder = append(seriesOrder, key) } series.Points = append(series.Points, HistorySeriesPoint{Time: historySeriesPointTime(bucket.Time), Value: spec.value(bucket), Min: spec.minimum(bucket), Max: spec.maximum(bucket), Count: bucket.Count}) } } } series := make([]HistorySeries, 0, len(seriesOrder)) returnedPoints := 0 for _, key := range seriesOrder { current := *seriesByKey[key] returnedPoints += len(current.Points) series = append(series, current) } windowBuckets := int(math.Ceil(end.Sub(start).Seconds() / float64(grainSeconds))) groupCount := len(groupBuckets) if groupCount == 0 { groupCount = len(vehicles) } expectedBuckets := windowBuckets * groupCount actualBuckets := 0 for _, count := range groupBuckets { actualBuckets += count } missingBuckets := expectedBuckets - actualBuckets if missingBuckets < 0 { missingBuckets = 0 } evidence := fmt.Sprintf("TDengine INTERVAL(%ds),按车辆与协议分区;空窗不补值", grainSeconds) return HistorySeriesResponse{Metrics: definitions, Series: series, DateFrom: start.UTC().Format(time.RFC3339), DateTo: end.UTC().Format(time.RFC3339), AsOf: time.Now().UTC().Format(time.RFC3339), Summary: HistorySeriesSummary{RawPointCount: rawPointCount, BucketCount: actualBuckets, ReturnedPointCount: returnedPoints, SeriesCount: len(series), GrainSeconds: grainSeconds, TargetPoints: targetPoints, ExpectedBucketCount: expectedBuckets, MissingBucketCount: missingBuckets, QueryDuration: time.Since(started).Milliseconds(), Complete: missingBuckets == 0, Evidence: evidence}}, nil } func historySeriesMetrics(raw string, specs map[string]historySeriesMetricSpec) []string { requested := strings.FieldsFunc(raw, func(char rune) bool { return char == ',' || char == ';' }) if len(requested) == 0 { requested = []string{"speedKmh", "totalMileageKm"} } result := make([]string, 0, len(requested)) seen := map[string]bool{} for _, key := range requested { key = strings.TrimSpace(key) if _, ok := specs[key]; ok && !seen[key] { result = append(result, key) seen[key] = true } } if len(result) == 0 { return []string{"speedKmh", "totalMileageKm"} } return result } func historySeriesWindow(rawFrom, rawTo string, now time.Time) (string, string, time.Time, time.Time, error) { shanghai := time.FixedZone("Asia/Shanghai", 8*60*60) localNow := now.In(shanghai) if strings.TrimSpace(rawFrom) == "" { rawFrom = time.Date(localNow.Year(), localNow.Month(), localNow.Day(), 0, 0, 0, 0, shanghai).Format("2006-01-02T15:04") } if strings.TrimSpace(rawTo) == "" { rawTo = localNow.Format("2006-01-02T15:04") } start, startOK := parseTrackRequestTime(rawFrom) end, endOK := parseTrackRequestTime(rawTo) if !startOK || !endOK { return "", "", time.Time{}, time.Time{}, clientError{Code: "HISTORY_TIME_INVALID", Message: "趋势时间格式无效"} } if !end.After(start) { return "", "", time.Time{}, time.Time{}, clientError{Code: "HISTORY_TIME_ORDER_INVALID", Message: "趋势结束时间必须晚于开始时间"} } if end.Sub(start) > 31*24*time.Hour { return "", "", time.Time{}, time.Time{}, clientError{Code: "HISTORY_TIME_RANGE_TOO_LARGE", Message: "单次趋势分析最长支持 31 天"} } return rawFrom, rawTo, start, end, nil } func historySeriesGrain(duration time.Duration, targetPoints int) int { required := int(math.Ceil(duration.Seconds() / float64(targetPoints))) for _, grain := range []int{1, 5, 10, 30, 60, 300, 900, 1800, 3600, 21600, 86400} { if required <= grain { return grain } } return 86400 } func historySeriesPointTime(value string) string { if parsed, ok := parseVehicleServiceTime(value); ok { return parsed.UTC().Format(time.RFC3339) } return value } func (s *Service) historySeriesPlate(ctx context.Context, vin string) string { page, err := s.store.Vehicles(ctx, url.Values{"keyword": {vin}, "limit": {"1"}}) if err == nil && len(page.Items) > 0 { return page.Items[0].Plate } return "" } func historyQueryKeywords(query url.Values) []string { raw := firstNonEmpty(query.Get("keywords"), query.Get("keyword"), query.Get("vin")) parts := strings.FieldsFunc(raw, func(char rune) bool { return char == ',' || char == ';' || char == '\n' }) result := make([]string, 0, len(parts)) seen := map[string]struct{}{} for _, part := range parts { part = strings.TrimSpace(part) if part == "" { continue } key := strings.ToLower(part) if _, ok := seen[key]; ok { continue } seen[key] = struct{}{} result = append(result, part) } return result } func (s *Service) historyDataForVehicle(ctx context.Context, category, keyword string, query url.Values, fetchLimit int) ([]HistoryDataRow, int, error) { protocol := strings.TrimSpace(query.Get("protocol")) switch category { case "location": resolved := url.Values{"keyword": {keyword}, "limit": {strconv.Itoa(fetchLimit)}, "offset": {"0"}} copyHistoryFilters(resolved, query, protocol) resolvedQuery, err := s.resolveVehicleQuery(ctx, resolved) if err != nil { return nil, 0, err } page, err := s.store.HistoryLocationsFromTDengine(ctx, resolvedQuery) if err != nil { return nil, 0, err } rows := make([]HistoryDataRow, 0, len(page.Items)) for index, item := range page.Items { quality := "normal" if !validTrackCoordinate(item) { quality = "warning" } rows = append(rows, HistoryDataRow{ID: "location-" + item.VIN + "-" + strconv.Itoa(index) + "-" + item.DeviceTime, VIN: item.VIN, Plate: item.Plate, Protocol: item.Protocol, DeviceTime: item.DeviceTime, ServerTime: item.ServerTime, Quality: quality, Values: map[string]any{"speedKmh": item.SpeedKmh, "totalMileageKm": item.TotalMileageKm, "longitude": item.Longitude, "latitude": item.Latitude}}) } return rows, page.Total, nil case "raw": resolvedVIN, err := s.resolveVehicleVIN(ctx, keyword, protocol) if err != nil { return nil, 0, err } page, err := s.store.RawFrames(ctx, RawFrameQuery{VIN: resolvedVIN, Protocol: protocol, DateFrom: query.Get("dateFrom"), DateTo: query.Get("dateTo"), IncludeFields: true, Limit: fetchLimit}) if err != nil { return nil, 0, err } rows := make([]HistoryDataRow, 0, len(page.Items)) for _, item := range page.Items { values := map[string]any{"frameType": item.FrameType, "rawSizeBytes": item.RawSizeBytes} for key, value := range item.ParsedFields { values[key] = value } rows = append(rows, HistoryDataRow{ID: item.ID, VIN: item.VIN, Plate: item.Plate, Protocol: item.Protocol, DeviceTime: item.DeviceTime, ServerTime: item.ServerTime, Quality: "normal", EvidenceID: item.ID, Values: values}) } return rows, page.Total, nil case "mileage": resolved := url.Values{"keyword": {keyword}, "limit": {strconv.Itoa(fetchLimit)}, "offset": {"0"}} copyHistoryFilters(resolved, query, protocol) resolvedQuery, err := s.resolveVehicleQuery(ctx, resolved) if err != nil { return nil, 0, err } page, err := s.store.DailyMileage(ctx, resolvedQuery) if err != nil { return nil, 0, err } rows := make([]HistoryDataRow, 0, len(page.Items)) for index, item := range page.Items { quality := "normal" if item.AnomalySeverity != "" { quality = item.AnomalySeverity } rows = append(rows, HistoryDataRow{ID: "mileage-" + item.VIN + "-" + strconv.Itoa(index) + "-" + item.Date, VIN: item.VIN, Plate: item.Plate, Protocol: item.Source, DeviceTime: item.Date, ServerTime: item.Date, Quality: quality, Values: map[string]any{"startMileageKm": item.StartMileageKm, "endMileageKm": item.EndMileageKm, "dailyMileageKm": item.DailyMileageKm}}) } return rows, page.Total, nil } return nil, 0, nil } func copyHistoryFilters(target, source url.Values, protocol string) { if protocol != "" { target.Set("protocol", protocol) } if value := strings.TrimSpace(source.Get("dateFrom")); value != "" { target.Set("dateFrom", value) } if value := strings.TrimSpace(source.Get("dateTo")); value != "" { target.Set("dateTo", value) } } func historyLocationMetrics() []HistoryMetricDefinition { return []HistoryMetricDefinition{ {Key: "speedKmh", Label: "速度", Unit: "km/h", Category: "location", ValueType: "number", DefaultVisible: true}, {Key: "totalMileageKm", Label: "总里程", Unit: "km", Category: "location", ValueType: "number", DefaultVisible: true}, {Key: "longitude", Label: "经度", Unit: "°", Category: "location", ValueType: "number", DefaultVisible: true}, {Key: "latitude", Label: "纬度", Unit: "°", Category: "location", ValueType: "number", DefaultVisible: true}, } } func historyRawMetrics() []HistoryMetricDefinition { return []HistoryMetricDefinition{ {Key: "frameType", Label: "报文类型", Category: "raw", ValueType: "string", DefaultVisible: true}, {Key: "rawSizeBytes", Label: "报文大小", Unit: "B", Category: "raw", ValueType: "number", DefaultVisible: true}, } } func historyMileageMetrics() []HistoryMetricDefinition { return []HistoryMetricDefinition{ {Key: "startMileageKm", Label: "起始里程", Unit: "km", Category: "mileage", ValueType: "number", DefaultVisible: true}, {Key: "endMileageKm", Label: "结束里程", Unit: "km", Category: "mileage", ValueType: "number", DefaultVisible: true}, {Key: "dailyMileageKm", Label: "日里程", Unit: "km", Category: "mileage", ValueType: "number", DefaultVisible: true}, } } func historyMetricsForCategory(category string) []HistoryMetricDefinition { switch category { case "raw": return historyRawMetrics() case "mileage": return historyMileageMetrics() default: return historyLocationMetrics() } } func mergeDiscoveredRawMetrics(base []HistoryMetricDefinition, rows []HistoryDataRow) []HistoryMetricDefinition { result := append([]HistoryMetricDefinition(nil), base...) seen := map[string]struct{}{} for _, metric := range result { seen[metric.Key] = struct{}{} } keys := make([]string, 0) for _, row := range rows { for key := range row.Values { if _, ok := seen[key]; !ok { seen[key] = struct{}{} keys = append(keys, key) } } } sort.Strings(keys) for _, key := range keys { label, unit := rawMetricMetadata(key) result = append(result, HistoryMetricDefinition{Key: key, Label: label, Unit: unit, Category: "raw", ValueType: "dynamic", DefaultVisible: len(result) < 7}) } return result } func rawMetricMetadata(key string) (string, string) { parts := strings.Split(key, ".") tail := parts[len(parts)-1] protocol := strings.ToUpper(parts[0]) labels := map[string]string{"speed_kmh": "速度", "total_mileage_km": "总里程", "avg_voltage_v": "平均电压", "soc_percent": "SOC", "longitude": "经度", "latitude": "纬度", "direction": "方向", "current_a": "电流", "temperature_c": "温度"} label := labels[tail] if label == "" { label = strings.ReplaceAll(tail, "_", " ") } if protocol != "" { label += " · " + protocol } unit := "" switch { case strings.HasSuffix(tail, "_kmh"): unit = "km/h" case strings.HasSuffix(tail, "_km"): unit = "km" case strings.HasSuffix(tail, "_percent"): unit = "%" case strings.HasSuffix(tail, "_voltage_v") || strings.HasSuffix(tail, "_v"): unit = "V" case strings.HasSuffix(tail, "_current_a") || strings.HasSuffix(tail, "_a"): unit = "A" case strings.HasSuffix(tail, "_temperature_c") || strings.HasSuffix(tail, "_c"): unit = "℃" } return label, unit } func (s *Service) CreateHistoryExport(request HistoryExportRequest) (HistoryExportJob, error) { request.Keywords = normalizedKeywords(request.Keywords) if len(request.Keywords) == 0 || len(request.Keywords) > 5 { return HistoryExportJob{}, clientError{Code: "EXPORT_SCOPE_INVALID", Message: "导出任务需要 1 至 5 台车辆"} } request.Format = strings.ToLower(strings.TrimSpace(request.Format)) if request.Format == "" { request.Format = "csv" } if request.Format != "csv" { return HistoryExportJob{}, clientError{Code: "EXPORT_FORMAT_INVALID", Message: "当前版本的受控导出仅支持 CSV"} } if request.Category == "" { request.Category = "location" } request.Category = strings.ToLower(strings.TrimSpace(request.Category)) if request.Category != "location" && request.Category != "raw" && request.Category != "mileage" { return HistoryExportJob{}, clientError{Code: "EXPORT_CATEGORY_INVALID", Message: "导出类型仅支持 location、raw 或 mileage"} } dateFrom, dateTo, _, _, windowErr := historySeriesWindow(request.DateFrom, request.DateTo, time.Now()) if windowErr != nil { return HistoryExportJob{}, windowErr } request.DateFrom, request.DateTo = dateFrom, dateTo request.Metrics = normalizedKeywords(request.Metrics) if len(request.Metrics) > 32 { return HistoryExportJob{}, clientError{Code: "EXPORT_METRIC_LIMIT_EXCEEDED", Message: "单次导出最多支持 32 个指标"} } identifier, err := randomExportID() if err != nil { return HistoryExportJob{}, err } now := time.Now().UTC().Format(time.RFC3339) job := &HistoryExportJob{ID: identifier, Name: "历史数据_" + time.Now().Format("20060102_150405"), Status: "queued", Progress: 0, Format: request.Format, Category: request.Category, Keywords: append([]string(nil), request.Keywords...), CreatedAt: now, UpdatedAt: now, Evidence: "单并发流式任务;最多 1,000,000 行;完成文件原子发布"} s.exportsMu.Lock() s.exports[job.ID] = job if err := s.persistHistoryExportsLocked(); err != nil { delete(s.exports, job.ID) s.exportsMu.Unlock() return HistoryExportJob{}, fmt.Errorf("persist export job: %w", err) } s.exportsMu.Unlock() go s.runHistoryExport(job.ID, request) return copyHistoryExportJob(job), nil } func (s *Service) ListHistoryExports() []HistoryExportJob { s.exportsMu.RLock() result := make([]HistoryExportJob, 0, len(s.exports)) for _, job := range s.exports { result = append(result, copyHistoryExportJob(job)) } s.exportsMu.RUnlock() sort.Slice(result, func(i, j int) bool { return result[i].CreatedAt > result[j].CreatedAt }) if len(result) > 20 { result = result[:20] } return result } func (s *Service) HistoryExportFile(id string) (string, string, error) { s.exportsMu.RLock() job := s.exports[id] if job == nil { s.exportsMu.RUnlock() return "", "", clientError{Code: "EXPORT_NOT_FOUND", Message: "导出任务不存在"} } copy := copyHistoryExportJob(job) path := job.filePath s.exportsMu.RUnlock() if copy.Status != "completed" || path == "" { return "", "", clientError{Code: "EXPORT_NOT_READY", Message: "导出文件尚未生成"} } return path, copy.Name + ".csv", nil } func (s *Service) runHistoryExport(id string, request HistoryExportRequest) { s.exportSlots <- struct{}{} defer func() { <-s.exportSlots }() s.updateHistoryExport(id, func(job *HistoryExportJob) { job.Status = "running"; job.Progress = 1 }) ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute) defer cancel() store, ok := s.store.(interface { HistoryExportCount(context.Context, HistoryExportStoreQuery) (int64, error) HistoryExportBatch(context.Context, HistoryExportStoreQuery, HistoryExportCursor, int) ([]HistoryDataRow, HistoryExportCursor, error) }) if !ok { s.failHistoryExport(id, errors.New("当前数据存储不支持流式历史导出")) return } resolvedVINs := make([]string, 0, len(request.Keywords)) seenVINs := map[string]bool{} for _, keyword := range request.Keywords { vin, err := s.resolveVehicleVIN(ctx, keyword, request.Protocol) if err != nil { s.failHistoryExport(id, err) return } if vin = strings.TrimSpace(vin); vin != "" && !seenVINs[vin] { seenVINs[vin] = true resolvedVINs = append(resolvedVINs, vin) } } queries := make([]HistoryExportStoreQuery, 0, len(resolvedVINs)) var totalRows int64 for _, vin := range resolvedVINs { query := HistoryExportStoreQuery{Category: request.Category, VIN: vin, Protocol: request.Protocol, DateFrom: request.DateFrom, DateTo: request.DateTo, Metrics: append([]string(nil), request.Metrics...)} count, err := store.HistoryExportCount(ctx, query) if err != nil { s.failHistoryExport(id, err) return } totalRows += count if totalRows > 1_000_000 { s.failHistoryExport(id, fmt.Errorf("结果 %d 行,超过单任务 1,000,000 行上限,请缩小车辆或时间范围", totalRows)) return } queries = append(queries, query) } s.updateHistoryExport(id, func(job *HistoryExportJob) { job.TotalRows = totalRows; job.Progress = 3 }) columns := historyExportColumns(request.Category, request.Metrics) if err := os.MkdirAll(s.exportDir, 0o750); err != nil { s.failHistoryExport(id, err) return } path := filepath.Join(s.exportDir, id+".csv") partPath := path + ".part" _ = os.Remove(partPath) file, err := os.OpenFile(partPath, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o640) if err != nil { s.failHistoryExport(id, err) return } completed := false defer func() { if !completed { _ = file.Close() _ = os.Remove(partPath) } }() if _, err := file.WriteString("\uFEFF"); err != nil { s.failHistoryExport(id, err) return } buffered := bufio.NewWriterSize(file, 1<<20) writer := csv.NewWriter(buffered) if err := writeHistoryExportMetadata(writer, request, columns); err != nil { s.failHistoryExport(id, err) return } var processed int64 for _, query := range queries { cursor := HistoryExportCursor{} for { rows, nextCursor, err := store.HistoryExportBatch(ctx, query, cursor, 5000) if err != nil { s.failHistoryExport(id, err) return } if len(rows) == 0 { break } if processed+int64(len(rows)) > 1_000_000 { s.failHistoryExport(id, errors.New("导出期间数据增长后超过 1,000,000 行上限,任务已安全停止")) return } for _, row := range rows { if err := writer.Write(historyExportRecord(row, columns)); err != nil { s.failHistoryExport(id, err) return } } writer.Flush() if err := writer.Error(); err != nil { s.failHistoryExport(id, err) return } if err := buffered.Flush(); err != nil { s.failHistoryExport(id, err) return } processed += int64(len(rows)) stat, _ := file.Stat() progress := 95 if totalRows > 0 { progress = 3 + int(float64(processed)/float64(totalRows)*92) if progress > 95 { progress = 95 } } s.updateHistoryExport(id, func(job *HistoryExportJob) { job.ProcessedRows = processed job.Progress = progress if stat != nil { job.FileSizeBytes = stat.Size() } }) if nextCursor == cursor { s.failHistoryExport(id, errors.New("导出游标未推进,任务已安全停止")) return } cursor = nextCursor } } writer.Flush() writeErr := writer.Error() bufferErr := buffered.Flush() syncErr := file.Sync() closeErr := file.Close() if writeErr != nil { s.failHistoryExport(id, writeErr) return } if bufferErr != nil { s.failHistoryExport(id, bufferErr) return } if syncErr != nil { s.failHistoryExport(id, syncErr) return } if closeErr != nil { s.failHistoryExport(id, closeErr) return } if err := os.Rename(partPath, path); err != nil { s.failHistoryExport(id, err) return } completed = true stat, _ := os.Stat(path) completedAt := time.Now().UTC().Format(time.RFC3339) s.updateHistoryExport(id, func(job *HistoryExportJob) { job.Status = "completed" job.Progress = 100 job.RowCount = int(processed) job.ProcessedRows = processed job.TotalRows = totalRows job.CompletedAt = completedAt if stat != nil { job.FileSizeBytes = stat.Size() } job.filePath = path job.DownloadURL = "/api/v2/exports/" + job.ID + "/download" }) } func (s *Service) updateHistoryExport(id string, update func(*HistoryExportJob)) { s.exportsMu.Lock() defer s.exportsMu.Unlock() if job := s.exports[id]; job != nil { update(job) job.UpdatedAt = time.Now().UTC().Format(time.RFC3339) _ = s.persistHistoryExportsLocked() } } func (s *Service) failHistoryExport(id string, err error) { s.updateHistoryExport(id, func(job *HistoryExportJob) { job.Status = "failed"; job.Error = err.Error(); job.Progress = 0 }) } func (s *Service) exportIndexPath() string { return filepath.Join(s.exportDir, "jobs.json") } func (s *Service) persistHistoryExportsLocked() error { if err := os.MkdirAll(s.exportDir, 0o750); err != nil { return err } jobs := make([]HistoryExportJob, 0, len(s.exports)) for _, job := range s.exports { jobs = append(jobs, copyHistoryExportJob(job)) } sort.Slice(jobs, func(i, j int) bool { return jobs[i].CreatedAt > jobs[j].CreatedAt }) if len(jobs) > 100 { jobs = jobs[:100] } contents, err := json.MarshalIndent(jobs, "", " ") if err != nil { return err } temporary := s.exportIndexPath() + ".tmp" if err := os.WriteFile(temporary, contents, 0o640); err != nil { return err } return os.Rename(temporary, s.exportIndexPath()) } func (s *Service) loadHistoryExports() { contents, err := os.ReadFile(s.exportIndexPath()) if err != nil { return } var jobs []HistoryExportJob if json.Unmarshal(contents, &jobs) != nil { return } now := time.Now().UTC().Format(time.RFC3339) for index := range jobs { job := jobs[index] if job.Status == "queued" || job.Status == "running" { job.Status = "failed" job.Progress = 0 job.Error = "服务重启,任务未完成,请重新创建导出" job.UpdatedAt = now } if job.Status == "completed" { job.filePath = filepath.Join(s.exportDir, job.ID+".csv") if _, err := os.Stat(job.filePath); err != nil { job.Status = "failed" job.Progress = 0 job.Error = "导出文件已不存在,请重新创建导出" job.DownloadURL = "" job.filePath = "" job.UpdatedAt = now } } s.exports[job.ID] = &job } s.exportsMu.Lock() _ = s.persistHistoryExportsLocked() s.exportsMu.Unlock() } func randomExportID() (string, error) { bytes := make([]byte, 8) if _, err := rand.Read(bytes); err != nil { return "", err } return "exp_" + hex.EncodeToString(bytes), nil } func copyHistoryExportJob(job *HistoryExportJob) HistoryExportJob { if job == nil { return HistoryExportJob{} } copy := *job copy.Keywords = append([]string(nil), job.Keywords...) return copy } func selectedExportMetrics(columns []HistoryMetricDefinition, requested []string) []HistoryMetricDefinition { if len(requested) == 0 { return append([]HistoryMetricDefinition(nil), columns...) } wanted := map[string]struct{}{} for _, key := range requested { wanted[key] = struct{}{} } result := make([]HistoryMetricDefinition, 0, len(requested)) for _, column := range columns { if _, ok := wanted[column.Key]; ok { result = append(result, column) } } return result } func historyExportColumns(category string, requested []string) []HistoryMetricDefinition { base := historyMetricsForCategory(category) selected := selectedExportMetrics(base, requested) if category != "raw" || len(requested) == 0 { return selected } known := map[string]bool{} for _, column := range selected { known[column.Key] = true } for _, key := range requested { if known[key] { continue } label, unit := rawMetricMetadata(key) selected = append(selected, HistoryMetricDefinition{Key: key, Label: label, Unit: unit, Category: "raw", ValueType: "dynamic", DefaultVisible: true}) known[key] = true } return selected } func writeHistoryExportMetadata(writer *csv.Writer, request HistoryExportRequest, columns []HistoryMetricDefinition) error { rows := [][]string{ {"导出元数据", "查询开始", request.DateFrom, "查询结束", request.DateTo, "车辆", strings.Join(request.Keywords, "、"), "数据类型", request.Category, "协议", firstNonEmpty(request.Protocol, "全部")}, {"指标与单位", strings.Join(exportMetricLabels(columns), ";")}, append([]string{"设备时间", "服务时间", "车牌", "VIN", "协议", "数据质量", "证据ID"}, exportMetricLabels(columns)...), } for _, row := range rows { if err := writer.Write(row); err != nil { return err } } return nil } func historyExportRecord(row HistoryDataRow, columns []HistoryMetricDefinition) []string { record := []string{row.DeviceTime, row.ServerTime, row.Plate, row.VIN, row.Protocol, row.Quality, row.EvidenceID} for _, column := range columns { value, ok := row.Values[column.Key] if !ok || value == nil { record = append(record, "") continue } record = append(record, fmt.Sprint(value)) } return record } func exportMetricLabels(columns []HistoryMetricDefinition) []string { labels := make([]string, 0, len(columns)) for _, column := range columns { label := column.Label if column.Unit != "" { label += "(" + column.Unit + ")" } labels = append(labels, label) } return labels } func summarizeTrack(points []HistoryLocationRow) TrackPlaybackSummary { summary := TrackPlaybackSummary{PointCount: len(points)} if len(points) == 0 { return summary } summary.StartTime = points[0].DeviceTime summary.EndTime = points[len(points)-1].DeviceTime var speedTotal float64 var speedCount int var coordinateDistance float64 for index, point := range points { if point.SpeedKmh >= 0 { speedTotal += point.SpeedKmh speedCount++ } if point.SpeedKmh > summary.MaximumSpeedKmh { summary.MaximumSpeedKmh = point.SpeedKmh } if index > 0 && validTrackCoordinate(points[index-1]) && validTrackCoordinate(point) { previousTime, previousOK := parseVehicleServiceTime(points[index-1].DeviceTime) currentTime, currentOK := parseVehicleServiceTime(point.DeviceTime) if previousOK && currentOK && currentTime.After(previousTime) && currentTime.Sub(previousTime) <= time.Duration(trackGapThresholdSeconds)*time.Second { coordinateDistance += haversineKm(points[index-1].Latitude, points[index-1].Longitude, point.Latitude, point.Longitude) } } } if speedCount > 0 { summary.AverageSpeedKmh = speedTotal / float64(speedCount) } mileageDistance := points[len(points)-1].TotalMileageKm - points[0].TotalMileageKm if points[0].Protocol == points[len(points)-1].Protocol && mileageDistance >= 0 && mileageDistance < 10000 { summary.DistanceKm = mileageDistance } else { summary.DistanceKm = coordinateDistance } start, startOK := parseVehicleServiceTime(summary.StartTime) end, endOK := parseVehicleServiceTime(summary.EndTime) if startOK && endOK && end.After(start) { summary.DurationSeconds = int64(end.Sub(start).Seconds()) } return summary } func summarizeTrackSources(points []HistoryLocationRow) []TrackPlaybackSource { byProtocol := map[string]*TrackPlaybackSource{} for _, point := range points { protocol := strings.TrimSpace(point.Protocol) if protocol == "" { protocol = "UNKNOWN" } source := byProtocol[protocol] if source == nil { source = &TrackPlaybackSource{Protocol: protocol, StartTime: point.DeviceTime} byProtocol[protocol] = source } source.PointCount++ source.EndTime = point.DeviceTime } keys := make([]string, 0, len(byProtocol)) for key := range byProtocol { keys = append(keys, key) } sort.Strings(keys) result := make([]TrackPlaybackSource, 0, len(keys)) for _, key := range keys { result = append(result, *byProtocol[key]) } return result } const ( trackGapThresholdSeconds = int64(10 * 60) trackStopMinimumSeconds = int64(3 * 60) trackStopSpeedKmh = 3.0 trackDriftSpeedKmh = 220.0 trackEventLimit = 64 ) func trackCoverage(query url.Values, total, fetched, processed, returned int, truncated, sampled bool) TrackCoverage { coverage := TrackCoverage{ RequestedStart: strings.TrimSpace(query.Get("dateFrom")), RequestedEnd: strings.TrimSpace(query.Get("dateTo")), TotalPoints: total, FetchedPoints: fetched, ProcessedPoints: processed, ReturnedPoints: returned, Complete: !truncated, LimitReasons: []string{}, } if truncated { coverage.LimitReasons = append(coverage.LimitReasons, "latest_source_slice") } if processed < fetched { coverage.LimitReasons = append(coverage.LimitReasons, "quality_filter") } if sampled { coverage.LimitReasons = append(coverage.LimitReasons, "map_sampling") } if truncated { coverage.Evidence = fmt.Sprintf("查询共 %d 个源点,仅加工最新 %d 个;摘要不代表完整时间窗", total, fetched) } else { coverage.Evidence = fmt.Sprintf("查询窗 %d 个源点已完整读取", total) } return coverage } func analyzeTrackPoints(raw []HistoryLocationRow) ([]HistoryLocationRow, TrackQuality) { quality := TrackQuality{RawPoints: len(raw)} points := make([]HistoryLocationRow, 0, len(raw)) lastByProtocol := map[string]HistoryLocationRow{} for _, point := range raw { if !validTrackCoordinate(point) { quality.InvalidCoordinatePoints++ continue } protocol := firstNonEmpty(strings.TrimSpace(point.Protocol), "UNKNOWN") if previous, ok := lastByProtocol[protocol]; ok { if point.DeviceTime == previous.DeviceTime && point.Protocol == previous.Protocol && math.Abs(point.Longitude-previous.Longitude) < 0.000001 && math.Abs(point.Latitude-previous.Latitude) < 0.000001 { quality.DuplicatePoints++ continue } previousTime, previousOK := parseVehicleServiceTime(previous.DeviceTime) currentTime, currentOK := parseVehicleServiceTime(point.DeviceTime) if previousOK && currentOK && currentTime.After(previousTime) { seconds := currentTime.Sub(previousTime).Seconds() distance := haversineKm(previous.Latitude, previous.Longitude, point.Latitude, point.Longitude) impliedSpeed := distance / seconds * 3600 if seconds <= 300 && distance >= 0.5 && impliedSpeed > trackDriftSpeedKmh { quality.DriftPoints++ continue } } } points = append(points, point) lastByProtocol[protocol] = point } quality.ValidPoints = len(points) issues := quality.InvalidCoordinatePoints + quality.DuplicatePoints + quality.DriftPoints if issues == 0 { quality.Status = "good" quality.Evidence = "坐标、重复点、漂移和时间间隔检查未发现异常" } else { quality.Status = "warning" quality.Evidence = fmt.Sprintf("过滤无效坐标 %d、重复 %d、疑似漂移 %d;检测到大间隔 %d", quality.InvalidCoordinatePoints, quality.DuplicatePoints, quality.DriftPoints, quality.LargeGapCount) } return points, quality } func selectTrackPrimarySource(points []HistoryLocationRow, requestedProtocol string) ([]HistoryLocationRow, string, int) { requestedProtocol = strings.ToUpper(strings.TrimSpace(requestedProtocol)) counts := map[string]int{} latest := map[string]time.Time{} for _, point := range points { protocol := strings.ToUpper(firstNonEmpty(strings.TrimSpace(point.Protocol), "UNKNOWN")) counts[protocol]++ if observedAt, ok := parseVehicleServiceTime(point.DeviceTime); ok && observedAt.After(latest[protocol]) { latest[protocol] = observedAt } } selected := requestedProtocol if selected == "" { protocols := make([]string, 0, len(counts)) for protocol := range counts { protocols = append(protocols, protocol) } sort.Slice(protocols, func(i, j int) bool { if counts[protocols[i]] != counts[protocols[j]] { return counts[protocols[i]] > counts[protocols[j]] } if !latest[protocols[i]].Equal(latest[protocols[j]]) { return latest[protocols[i]].After(latest[protocols[j]]) } return protocols[i] < protocols[j] }) if len(protocols) > 0 { selected = protocols[0] } } result := make([]HistoryLocationRow, 0, counts[selected]) for _, point := range points { if strings.EqualFold(firstNonEmpty(strings.TrimSpace(point.Protocol), "UNKNOWN"), selected) { result = append(result, point) } } return result, selected, len(points) - len(result) } func applyTrackSequenceQuality(points []HistoryLocationRow, quality *TrackQuality) { quality.SourceSwitches = 0 quality.LargeGapCount = 0 quality.MaximumGapSeconds = 0 for index := 1; index < len(points); index++ { if points[index].Protocol != points[index-1].Protocol { quality.SourceSwitches++ } previousTime, previousOK := parseVehicleServiceTime(points[index-1].DeviceTime) currentTime, currentOK := parseVehicleServiceTime(points[index].DeviceTime) if !previousOK || !currentOK || !currentTime.After(previousTime) { continue } gap := int64(currentTime.Sub(previousTime).Seconds()) if gap > quality.MaximumGapSeconds { quality.MaximumGapSeconds = gap } if gap > trackGapThresholdSeconds { quality.LargeGapCount++ } } issues := quality.InvalidCoordinatePoints + quality.DuplicatePoints + quality.DriftPoints + quality.LargeGapCount if issues == 0 { quality.Status = "good" quality.Evidence = "主来源坐标、重复点、漂移和时间间隔检查未发现异常" } else { quality.Status = "warning" quality.Evidence = fmt.Sprintf("主来源过滤无效坐标 %d、重复 %d、疑似漂移 %d;检测到大间隔 %d", quality.InvalidCoordinatePoints, quality.DuplicatePoints, quality.DriftPoints, quality.LargeGapCount) } } func applyTrackPrimarySourceCoverage(coverage *TrackCoverage, protocol string, alternatePoints int) { if alternatePoints <= 0 { return } coverage.LimitReasons = append(coverage.LimitReasons, "primary_source_projection") coverage.Evidence += fmt.Sprintf(";地图选用 %s 主来源,另 %d 个有效点保留在来源统计", firstNonEmpty(protocol, "UNKNOWN"), alternatePoints) } func buildTrackSegments(points []HistoryLocationRow) []TrackSegment { if len(points) == 0 { return []TrackSegment{} } if len(points) == 1 { return []TrackSegment{{Index: 0, Type: "point", Title: "单点", StartTime: points[0].DeviceTime, EndTime: points[0].DeviceTime, PointCount: 1}} } segments := make([]TrackSegment, 0) for index := 1; index < len(points); index++ { previous, current := points[index-1], points[index] previousTime, previousOK := parseVehicleServiceTime(previous.DeviceTime) currentTime, currentOK := parseVehicleServiceTime(current.DeviceTime) duration := int64(0) validInterval := previousOK && currentOK && currentTime.After(previousTime) if validInterval { duration = int64(currentTime.Sub(previousTime).Seconds()) } distance := haversineKm(previous.Latitude, previous.Longitude, current.Latitude, current.Longitude) typeName, title := "moving", "行驶" switch { case !validInterval || currentTime.Sub(previousTime) > time.Duration(trackGapThresholdSeconds)*time.Second: typeName, title = "gap", "数据间隔" case previous.SpeedKmh <= trackStopSpeedKmh && current.SpeedKmh <= trackStopSpeedKmh && distance <= 0.15: typeName, title = "stopped", "停车(GPS 推断)" } if typeName == "gap" { distance = 0 } if len(segments) > 0 && segments[len(segments)-1].Type == typeName && segments[len(segments)-1].EndIndex == index-1 { segment := &segments[len(segments)-1] segment.EndIndex = index segment.EndTime = current.DeviceTime segment.DurationSeconds += duration segment.DistanceKm += distance segment.PointCount++ continue } segments = append(segments, TrackSegment{ Index: len(segments), Type: typeName, Title: title, StartTime: previous.DeviceTime, EndTime: current.DeviceTime, DurationSeconds: duration, DistanceKm: distance, PointCount: 2, StartIndex: index - 1, EndIndex: index, }) } return segments } func buildTrackStops(points []HistoryLocationRow, segments []TrackSegment) []TrackStop { stops := make([]TrackStop, 0) for _, segment := range segments { if segment.Type != "stopped" || segment.DurationSeconds < trackStopMinimumSeconds { continue } longitude, latitude := 0.0, 0.0 count := 0 for index := segment.StartIndex; index <= segment.EndIndex && index < len(points); index++ { longitude += points[index].Longitude latitude += points[index].Latitude count++ } if count > 0 { longitude /= float64(count) latitude /= float64(count) } stops = append(stops, TrackStop{ Index: len(stops), StartTime: segment.StartTime, EndTime: segment.EndTime, DurationSeconds: segment.DurationSeconds, PointCount: segment.PointCount, Longitude: longitude, Latitude: latitude, Evidence: "由连续低速且小位移位置点推断,不代表点火状态", startIndex: segment.StartIndex, endIndex: segment.EndIndex, }) } return stops } func buildTrackEvents(points []HistoryLocationRow, segments []TrackSegment, stops []TrackStop) []TrackPlaybackEvent { if len(points) == 0 { return []TrackPlaybackEvent{} } events := []TrackPlaybackEvent{trackEvent(0, "start", "开始行驶", points[0])} for _, stop := range stops { events = append(events, trackEvent(stop.startIndex, "stop", "停车(GPS 推断)", points[stop.startIndex])) } for _, segment := range segments { if segment.Type == "gap" && segment.EndIndex < len(points) { events = append(events, trackEvent(segment.EndIndex, "gap", "数据间隔", points[segment.EndIndex])) } } for index := 1; index < len(points)-1; index++ { delta := points[index].SpeedKmh - points[index-1].SpeedKmh typeName, title := "", "" switch { case points[index].Protocol != points[index-1].Protocol: typeName, title = "source_switch", "数据源切换" case delta >= 25: typeName, title = "acceleration", "急加速" case delta <= -25: typeName, title = "braking", "急减速" } if typeName != "" { events = append(events, trackEvent(index, typeName, title, points[index])) } } events = append(events, trackEvent(len(points)-1, "end", "结束行驶", points[len(points)-1])) sort.SliceStable(events, func(i, j int) bool { return events[i].Index < events[j].Index }) if len(events) > trackEventLimit { limited := append([]TrackPlaybackEvent(nil), events[:trackEventLimit-1]...) limited = append(limited, events[len(events)-1]) events = limited } return events } func trackEvent(index int, eventType, title string, point HistoryLocationRow) TrackPlaybackEvent { return TrackPlaybackEvent{Index: index, Type: eventType, Title: title, Time: point.DeviceTime, SpeedKmh: point.SpeedKmh, SOCPercent: point.SOCPercent, SOCAvailable: point.SOCAvailable, DirectionDeg: point.DirectionDeg, AlarmFlag: point.AlarmFlag, Longitude: point.Longitude, Latitude: point.Latitude} } func trackKeyIndexes(events []TrackPlaybackEvent, segments []TrackSegment, stops []TrackStop) []int { indexes := make([]int, 0, len(events)+len(segments)*2+len(stops)) for _, event := range events { indexes = append(indexes, event.Index) } for _, segment := range segments { indexes = append(indexes, segment.StartIndex, segment.EndIndex) } for _, stop := range stops { indexes = append(indexes, (stop.startIndex+stop.endIndex)/2) } return indexes } func sampleTrackPoints(points []HistoryLocationRow, maxPoints int) []HistoryLocationRow { result, _ := sampleTrackPointsWithKeys(points, maxPoints, nil) return result } func sampleTrackPointsWithKeys(points []HistoryLocationRow, maxPoints int, keyIndexes []int) ([]HistoryLocationRow, []int) { if len(points) <= maxPoints { indexes := make([]int, len(points)) for index := range points { indexes[index] = index } return append([]HistoryLocationRow(nil), points...), indexes } if maxPoints < 2 { maxPoints = 2 } selected := map[int]struct{}{0: {}, len(points) - 1: {}} for _, index := range keyIndexes { if index >= 0 && index < len(points) { selected[index] = struct{}{} } } indexes := sortedTrackIndexes(selected) if len(indexes) > maxPoints { indexes = uniformlySelectTrackIndexes(indexes, maxPoints) selected = make(map[int]struct{}, len(indexes)) for _, index := range indexes { selected[index] = struct{}{} } } remaining := maxPoints - len(selected) for slot := 1; slot <= remaining; slot++ { index := int(math.Round(float64(slot) * float64(len(points)-1) / float64(remaining+1))) selected[index] = struct{}{} } for len(selected) < maxPoints { indexes = sortedTrackIndexes(selected) largestGap, midpoint := 0, -1 for index := 1; index < len(indexes); index++ { gap := indexes[index] - indexes[index-1] if gap > largestGap { largestGap = gap midpoint = indexes[index-1] + gap/2 } } if midpoint < 0 || largestGap <= 1 { break } selected[midpoint] = struct{}{} } indexes = sortedTrackIndexes(selected) if len(indexes) > maxPoints { indexes = uniformlySelectTrackIndexes(indexes, maxPoints) } result := make([]HistoryLocationRow, 0, len(indexes)) for _, index := range indexes { result = append(result, points[index]) } return result, indexes } func sortedTrackIndexes(selected map[int]struct{}) []int { indexes := make([]int, 0, len(selected)) for index := range selected { indexes = append(indexes, index) } sort.Ints(indexes) return indexes } func uniformlySelectTrackIndexes(indexes []int, limit int) []int { if len(indexes) <= limit { return append([]int(nil), indexes...) } result := make([]int, 0, limit) for slot := 0; slot < limit; slot++ { index := int(math.Round(float64(slot) * float64(len(indexes)-1) / float64(limit-1))) result = append(result, indexes[index]) } return result } func nearestSampledIndex(original int, sampled []int) int { if len(sampled) == 0 { return 0 } position := sort.SearchInts(sampled, original) if position == 0 { return 0 } if position == len(sampled) { return len(sampled) - 1 } if original-sampled[position-1] <= sampled[position]-original { return position - 1 } return position } func validTrackCoordinate(point HistoryLocationRow) bool { return point.Longitude >= 72 && point.Longitude <= 138 && point.Latitude >= 0.8 && point.Latitude <= 56 } func haversineKm(lat1, lon1, lat2, lon2 float64) float64 { const earthRadiusKm = 6371.0088 toRadians := math.Pi / 180 dLat := (lat2 - lat1) * toRadians dLon := (lon2 - lon1) * toRadians a := math.Sin(dLat/2)*math.Sin(dLat/2) + math.Cos(lat1*toRadians)*math.Cos(lat2*toRadians)*math.Sin(dLon/2)*math.Sin(dLon/2) return earthRadiusKm * 2 * math.Atan2(math.Sqrt(a), math.Sqrt(1-a)) } func (s *Service) RawFrames(ctx context.Context, query RawFrameQuery) (Page[RawFrameRow], error) { if query.Limit <= 0 || query.Limit > 500 { query.Limit = 100 } if query.IncludeFields && !rawFrameQueryHasNarrowScope(query) { return Page[RawFrameRow]{}, clientError{ Code: "RAW_FRAME_SCOPE_REQUIRED", Message: "RAW 解析字段查询需要提供车辆或时间范围", } } if strings.TrimSpace(query.VIN) == "" { query.VIN = query.Keyword } resolvedVIN, err := s.resolveVehicleVIN(ctx, query.VIN, query.Protocol) if err != nil { return Page[RawFrameRow]{}, err } if resolvedVIN != "" { query.VIN = resolvedVIN } return s.store.RawFrames(ctx, query) } func rawFrameQueryHasNarrowScope(query RawFrameQuery) bool { return strings.TrimSpace(query.VIN) != "" || strings.TrimSpace(query.Keyword) != "" || strings.TrimSpace(query.DateFrom) != "" || strings.TrimSpace(query.DateTo) != "" || len(query.Fields) > 0 } func (s *Service) MileageSummary(ctx context.Context, query url.Values) (MileageSummary, error) { resolvedQuery, err := s.resolveVehicleQuery(ctx, query) if err != nil { return MileageSummary{}, err } return s.store.MileageSummary(ctx, resolvedQuery) } func (s *Service) DailyMileage(ctx context.Context, query url.Values) (Page[DailyMileageRow], error) { resolvedQuery, err := s.resolveVehicleQuery(ctx, query) if err != nil { return Page[DailyMileageRow]{}, err } resolvedQuery, err = normalizeMileageVINSelection(resolvedQuery) if err != nil { return Page[DailyMileageRow]{}, err } resolvedQuery, err = normalizeDailyMileageWindow(resolvedQuery, time.Now()) if err != nil { return Page[DailyMileageRow]{}, err } return s.store.DailyMileage(ctx, resolvedQuery) } func (s *Service) MileageStatistics(ctx context.Context, query url.Values) (MileageStatistics, error) { resolvedQuery, err := s.resolveVehicleQuery(ctx, query) if err != nil { return MileageStatistics{}, err } resolvedQuery, err = normalizeMileageVINSelection(resolvedQuery) if err != nil { return MileageStatistics{}, err } resolvedQuery, err = normalizeMileageStatisticsWindow(resolvedQuery, time.Now()) if err != nil { return MileageStatistics{}, err } return s.store.MileageStatistics(ctx, resolvedQuery) } func normalizeMileageVINSelection(query url.Values) (url.Values, error) { next := cloneValues(query) seen := map[string]bool{} vins := make([]string, 0) for _, item := range strings.Split(next.Get("vins"), ",") { vin := strings.TrimSpace(item) if vin == "" || seen[vin] { continue } seen[vin] = true vins = append(vins, vin) } if len(vins) > 20 { return nil, clientError{Code: "TOO_MANY_VEHICLES", Message: "里程查询一次最多选择 20 辆车"} } if len(vins) == 0 { next.Del("vins") } else { next.Set("vins", strings.Join(vins, ",")) } return next, nil } func normalizeMileageStatisticsWindow(query url.Values, now time.Time) (url.Values, error) { next := cloneValues(query) dateTo := strings.TrimSpace(next.Get("dateTo")) if dateTo == "" { dateTo = now.Format("2006-01-02") next.Set("dateTo", dateTo) } dateFrom := strings.TrimSpace(next.Get("dateFrom")) if dateFrom == "" { parsedTo, _ := time.ParseInLocation("2006-01-02", dateTo, now.Location()) dateFrom = parsedTo.AddDate(0, 0, -29).Format("2006-01-02") next.Set("dateFrom", dateFrom) } from, fromErr := time.ParseInLocation("2006-01-02", dateFrom, now.Location()) to, toErr := time.ParseInLocation("2006-01-02", dateTo, now.Location()) if fromErr != nil || toErr != nil || from.After(to) { return nil, clientError{Code: "INVALID_DATE_RANGE", Message: "统计日期范围无效"} } if to.Sub(from) > 365*24*time.Hour { return nil, clientError{Code: "DATE_RANGE_TOO_LARGE", Message: "车辆统计最多查询 366 个自然日"} } return next, nil } func normalizeDailyMileageWindow(query url.Values, now time.Time) (url.Values, error) { if strings.TrimSpace(query.Get("dateFrom")) == "" && strings.TrimSpace(query.Get("dateTo")) == "" { return cloneValues(query), nil } return normalizeMileageStatisticsWindow(query, now) } func (s *Service) OnlineStatisticsSummary(ctx context.Context, query url.Values) (OnlineStatisticsSummary, error) { resolvedQuery, err := s.resolveVehicleQuery(ctx, query) if err != nil { return OnlineStatisticsSummary{}, err } coverage, err := s.store.VehicleCoverageSummary(ctx, resolvedQuery) if err != nil { return OnlineStatisticsSummary{}, err } serviceSummary, err := s.store.VehicleServiceSummary(ctx) if err != nil { return OnlineStatisticsSummary{}, err } opsHealth, err := s.store.OpsHealth(ctx) if err != nil { return OnlineStatisticsSummary{}, err } total := coverage.TotalVehicles online := coverage.OnlineVehicles offline := total - online if offline < 0 { offline = 0 } rate := 0.0 if total > 0 { rate = float64(online) / float64(total) * 100 } return OnlineStatisticsSummary{ VehicleCount: total, OnlineVehicleCount: online, OfflineVehicleCount: offline, OnlineRatePercent: rate, RedisOnlineKeys: opsHealth.RedisOnlineKeys, ProtocolStats: serviceSummary.Protocols, Evidence: "由车辆服务覆盖汇总、实时来源状态和 Redis 在线 key 计算", }, nil } func (s *Service) OnlineVehicleStatuses(ctx context.Context, query url.Values) (Page[OnlineVehicleStatusRow], error) { resolvedQuery, err := s.resolveVehicleQuery(ctx, query) if err != nil { return Page[OnlineVehicleStatusRow]{}, err } coverage, err := s.store.VehicleCoverage(ctx, resolvedQuery) if err != nil { return Page[OnlineVehicleStatusRow]{}, err } now := time.Now() items := make([]OnlineVehicleStatusRow, 0, len(coverage.Items)) for _, row := range coverage.Items { items = append(items, OnlineVehicleStatusRow{ VIN: row.VIN, Plate: row.Plate, Phone: row.Phone, OEM: row.OEM, Protocols: append([]string(nil), row.Protocols...), Online: row.Online, LastSeen: row.LastSeen, OfflineDurationMinutes: offlineDurationMinutes(row.Online, row.LastSeen, now), SourceCount: row.SourceCount, OnlineSourceCount: row.OnlineSourceCount, BindingStatus: row.BindingStatus, }) } return Page[OnlineVehicleStatusRow]{ Items: items, Total: coverage.Total, Limit: coverage.Limit, Offset: coverage.Offset, }, nil } func offlineDurationMinutes(online bool, lastSeen string, now time.Time) *int { if online { value := 0 return &value } parsed, ok := parseVehicleServiceTime(lastSeen) if !ok { return nil } minutes := int(now.Sub(parsed).Minutes()) if minutes < 0 { minutes = 0 } return &minutes } func (s *Service) QualitySummary(ctx context.Context, query url.Values) (QualitySummary, error) { return s.store.QualitySummary(ctx, query) } func (s *Service) QualityIssues(ctx context.Context, query url.Values) (Page[QualityIssueRow], error) { return s.store.QualityIssues(ctx, query) } func (s *Service) QualityNotificationPlan(ctx context.Context, query url.Values) (QualityNotificationPlan, error) { summary, err := s.store.QualitySummary(ctx, query) if err != nil { return QualityNotificationPlan{}, err } issueQuery := cloneValues(query) if issueQuery.Get("limit") == "" { issueQuery.Set("limit", "20") } issues, err := s.store.QualityIssues(ctx, issueQuery) if err != nil { return QualityNotificationPlan{}, err } health, err := s.store.OpsHealth(ctx) if err != nil { return QualityNotificationPlan{}, err } health.Runtime = s.runtime rules := qualityNotificationRules(summary, health) activeRuleCount := 0 p0RuleCount := 0 for _, rule := range rules { if rule.Count <= 0 { continue } activeRuleCount++ if rule.Level == "P0" { p0RuleCount++ } } return QualityNotificationPlan{ Summary: summary, Rules: rules, Policies: append([]QualityNotificationPolicy(nil), qualityNotificationPolicies...), PriorityIssues: qualityPriorityIssues(issues.Items), ActiveRuleCount: activeRuleCount, P0RuleCount: p0RuleCount, }, nil } func (s *Service) OpsHealth(ctx context.Context) (OpsHealth, error) { health, err := s.store.OpsHealth(ctx) if err != nil { return OpsHealth{}, err } health.Runtime = s.runtime return health, nil } func qualityNotificationRules(summary QualitySummary, health OpsHealth) []QualityAlertRule { rows := make([]QualityAlertRule, 0, len(qualityAlertRuleTemplates)+1) for _, item := range qualityAlertRuleTemplates { item.Count = qualityIssueCount(summary, item.IssueType) rows = append(rows, item) } storageWritable := health.TDengineWritable && health.MySQLWritable capacityCount := len(health.CapacityFindings) capacityRule := QualityAlertRule{ IssueType: "CAPACITY_RISK", Title: "容量与存储风险", Level: "P2", Owner: "基础设施", Trigger: "Kafka Lag、连接容量、Redis 在线 Key 或存储读写异常", Notify: "容量风险进入运维日报,超过阈值升级", SLA: "当日评估", Count: capacityCount, } if !storageWritable { capacityRule.Level = "P0" capacityRule.Notify = "立即通知基础设施值班人处理存储不可写" capacityRule.SLA = "15 分钟恢复" capacityRule.Count++ } rows = append(rows, capacityRule) return rows } func qualityIssueCount(summary QualitySummary, issueType string) int { for _, item := range summary.IssueTypes { if item.Name == issueType { return item.Count } } return 0 } func qualityPriorityIssues(rows []QualityIssueRow) []QualityPriorityIssue { out := make([]QualityPriorityIssue, 0, len(rows)) for _, row := range rows { out = append(out, buildQualityPriorityIssue(row)) } sort.Slice(out, func(i, j int) bool { if out[i].Priority != out[j].Priority { return out[i].Priority == "P0" } weightDelta := qualityIssuePriorityWeight[out[j].IssueType] - qualityIssuePriorityWeight[out[i].IssueType] if weightDelta != 0 { return weightDelta < 0 } return out[i].LastSeen > out[j].LastSeen }) if len(out) > 5 { out = out[:5] } return out } func buildQualityPriorityIssue(row QualityIssueRow) QualityPriorityIssue { actionLabel, actionDetail := qualityAction(row) priority := "P1" if row.Severity == "error" || row.IssueType == "NO_SOURCE" { priority = "P0" } lookup := qualityIssueLookup(row) dateFrom := evidenceDate(row.LastSeen) dateTo := nextEvidenceDate(dateFrom) realtimeHash := buildHash("realtime", lookup, row.Protocol, nil) historyFilters := map[string]string{} if dateFrom != "" { historyFilters["dateFrom"] = dateFrom } if dateTo != "" { historyFilters["dateTo"] = dateTo } historyHash := buildHash("history", lookup, row.Protocol, historyFilters) rawFilters := cloneStringMap(historyFilters) rawFilters["tab"] = "raw" rawFilters["includeFields"] = "true" rawHash := buildHash("history-query", lookup, row.Protocol, rawFilters) vehicleHash := buildHash("detail", lookup, row.Protocol, nil) item := QualityPriorityIssue{ QualityIssueRow: row, Priority: priority, ActionLabel: actionLabel, ActionDetail: actionDetail, SLA: qualityIssueSLA(row.IssueType), VehicleLabel: qualityVehicleLabel(row), RealtimeHash: realtimeHash, HistoryHash: historyHash, RawHash: rawHash, VehicleHash: vehicleHash, } item.NotificationText = qualityNotificationText(item) return item } func qualityAction(row QualityIssueRow) (string, string) { switch { case row.IssueType == "NO_SOURCE": return "确认平台转发", "车辆已绑定但没有任何来源证据,先确认平台转发、端口和订阅。" case row.IssueType == "VIN_MISSING" || strings.TrimSpace(row.VIN) == "": return "维护身份绑定", "数据已有来源但无法归并到 VIN,优先用车牌/手机号补齐绑定。" case row.IssueType == "LINK_GAP": return "排查来源链路", "来源存在上报间断,优先检查平台转发、网络和消费延迟。" case row.IssueType == "FIELD_MISSING": return "核对解析字段", "字段缺失会影响统计和展示,优先核对解析映射和原始 RAW。" default: return "查看车辆服务", "进入车辆服务详情,结合来源证据继续排查。" } } func qualityIssueSLA(issueType string) string { for _, item := range qualityAlertRuleTemplates { if item.IssueType == issueType { return item.SLA } } return "当日闭环" } func qualityVehicleLabel(row QualityIssueRow) string { identity := firstNonEmpty(strings.TrimSpace(row.VIN), strings.TrimSpace(row.Phone), strings.TrimSpace(row.SourceEndpoint), "-") if plate := strings.TrimSpace(row.Plate); plate != "" { return plate + " / " + identity } return identity } func qualityIssueLookup(row QualityIssueRow) string { return firstNonEmpty(strings.TrimSpace(row.VIN), strings.TrimSpace(row.Plate), strings.TrimSpace(row.Phone), strings.TrimSpace(row.SourceEndpoint)) } func evidenceDate(value string) string { value = strings.TrimSpace(value) if len(value) < len("2006-01-02") { return "" } date := value[:len("2006-01-02")] if _, err := time.Parse("2006-01-02", date); err != nil { return "" } return date } func nextEvidenceDate(value string) string { if value == "" { return "" } date, err := time.Parse("2006-01-02", value) if err != nil { return "" } return date.AddDate(0, 0, 1).Format("2006-01-02") } func buildHash(page string, keyword string, protocol string, filters map[string]string) string { values := url.Values{} if keyword != "" { values.Set("keyword", keyword) } if protocol != "" { values.Set("protocol", protocol) } for key, value := range filters { if strings.TrimSpace(value) != "" { values.Set(key, value) } } query := values.Encode() if query == "" { return "#/" + page } return "#/" + page + "?" + query } func qualityNotificationText(row QualityPriorityIssue) string { return strings.Join([]string{ "【" + row.Priority + " 告警通知】" + qualityIssueTitle(row.IssueType), "车辆:" + row.VehicleLabel, "数据来源:" + row.Protocol, "问题:" + qualityIssueTitle(row.IssueType), "建议动作:" + row.ActionLabel, "SLA:" + row.SLA, "最后时间:" + firstNonEmpty(row.LastSeen, "-"), "详情:" + firstNonEmpty(row.Detail, "-"), "实时定位:" + row.RealtimeHash, "轨迹证据:" + row.HistoryHash, "原始记录:" + row.RawHash, "车辆服务:" + row.VehicleHash, }, "\n") } func qualityIssueTitle(issueType string) string { for _, item := range qualityAlertRuleTemplates { if item.IssueType == issueType { return item.Title } } if issueType == "CAPACITY_RISK" { return "容量与存储风险" } return issueType } func cloneStringMap(values map[string]string) map[string]string { out := make(map[string]string, len(values)) for key, value := range values { out[key] = value } return out } func (s *Service) resolveVehicleQuery(ctx context.Context, query url.Values) (url.Values, error) { resolved := cloneValues(query) if strings.TrimSpace(resolved.Get("keywords")) != "" { return resolved, nil } vin := firstNonEmpty(resolved.Get("vin"), resolved.Get("keyword")) if vin == "" { return resolved, nil } resolvedVIN, err := s.resolveVehicleVIN(ctx, vin, resolved.Get("protocol")) if err != nil { return nil, err } if resolvedVIN != "" { resolved.Set("vin", resolvedVIN) } return resolved, nil } func (s *Service) resolveVehicleVIN(ctx context.Context, keyword string, protocol string) (string, error) { keyword = strings.TrimSpace(keyword) if keyword == "" || isLikelyVIN(keyword) { return keyword, nil } vehicleQuery := url.Values{"keyword": {keyword}, "limit": {"10"}} if protocol = strings.TrimSpace(protocol); protocol != "" { vehicleQuery.Set("protocol", protocol) } vehicles, err := s.store.Vehicles(ctx, vehicleQuery) if err != nil { return "", err } identity := resolveVehicleIdentity(keyword, vehicles.Items) if identity == nil || strings.TrimSpace(identity.VIN) == "" { return keyword, nil } return identity.VIN, nil } func cloneValues(values url.Values) url.Values { cloned := make(url.Values, len(values)) for key, current := range values { cloned[key] = append([]string(nil), current...) } return cloned } func normalizedKeywords(values []string) []string { out := make([]string, 0, len(values)) seen := map[string]struct{}{} for _, value := range values { value = strings.TrimSpace(value) if value == "" { continue } if _, ok := seen[value]; ok { continue } seen[value] = struct{}{} out = append(out, value) } return out } func overviewBatchPage(query VehicleOverviewBatchQuery) ([]string, int, int, int) { keywords := normalizedKeywords(query.Keywords) total := len(keywords) limit := query.Limit offset := query.Offset if offset < 0 { offset = 0 } if limit <= 0 || limit > 200 { limit = 200 } if offset >= total { return []string{}, total, limit, offset } end := offset + limit if end > total { end = total } return keywords[offset:end], total, limit, offset } func vehicleSourceStatus(vehicles []VehicleRow, realtime []RealtimeLocationRow, history []HistoryLocationRow, raw []RawFrameRow, mileage []DailyMileageRow) []VehicleSourceStatus { statusByProtocol := map[string]*VehicleSourceStatus{} ensure := func(protocol string) *VehicleSourceStatus { protocol = strings.TrimSpace(protocol) if protocol == "" { return nil } if current := statusByProtocol[protocol]; current != nil { return current } current := &VehicleSourceStatus{Protocol: protocol} statusByProtocol[protocol] = current return current } setLastSeen := func(current *VehicleSourceStatus, value string) { value = strings.TrimSpace(value) if value == "" { return } if current.LastSeen == "" || value > current.LastSeen { current.LastSeen = value } } for _, row := range vehicles { if current := ensure(row.Protocol); current != nil { current.Online = current.Online || row.Online setLastSeen(current, row.LastSeen) } } for _, row := range realtime { if current := ensure(row.Protocol); current != nil { current.HasRealtime = true setLastSeen(current, row.LastSeen) } } for _, row := range history { if current := ensure(row.Protocol); current != nil { current.HasHistory = true setLastSeen(current, firstNonEmpty(row.ServerTime, row.DeviceTime)) } } for _, row := range raw { if current := ensure(row.Protocol); current != nil { current.HasRaw = true setLastSeen(current, firstNonEmpty(row.ServerTime, row.DeviceTime)) } } for _, row := range mileage { if current := ensure(row.Source); current != nil { current.HasMileage = true setLastSeen(current, row.Date) } } out := make([]VehicleSourceStatus, 0, len(statusByProtocol)) for _, current := range statusByProtocol { out = append(out, *current) } sort.Slice(out, func(i, j int) bool { return out[i].Protocol < out[j].Protocol }) return out } func completeVehicleSourceStatus(statuses []VehicleSourceStatus, protocol string) []VehicleSourceStatus { protocol = strings.TrimSpace(protocol) if protocol != "" { for _, status := range statuses { if status.Protocol == protocol { return []VehicleSourceStatus{status} } } return []VehicleSourceStatus{{Protocol: protocol}} } byProtocol := map[string]VehicleSourceStatus{} for _, status := range statuses { if strings.TrimSpace(status.Protocol) == "" { continue } byProtocol[status.Protocol] = status } out := make([]VehicleSourceStatus, 0, len(canonicalVehicleProtocols)+len(statuses)) seen := map[string]struct{}{} for _, protocol := range canonicalVehicleProtocols { if status, ok := byProtocol[protocol]; ok { out = append(out, status) } else { out = append(out, VehicleSourceStatus{Protocol: protocol}) } seen[protocol] = struct{}{} } for _, status := range statuses { if _, ok := seen[status.Protocol]; ok || strings.TrimSpace(status.Protocol) == "" { continue } out = append(out, status) } return out } func buildVehicleServiceStatus(resolved bool, statuses []VehicleSourceStatus) *VehicleServiceStatus { if !resolved { return &VehicleServiceStatus{ Status: "identity_required", Severity: "warning", Title: "身份未绑定", Detail: "车辆关键词暂未解析到 VIN,需先维护身份绑定后才能形成完整车辆服务。", } } sourceCount := len(statuses) onlineSourceCount := 0 for _, status := range statuses { if status.Online { onlineSourceCount++ } } missingProtocols := missingProtocolsFromStatuses(statuses) if sourceCount == 0 || sourceCount == len(missingProtocols) { return &VehicleServiceStatus{ Status: "no_data", Severity: "warning", Title: "暂无数据来源", Detail: "车辆已解析,但暂未查询到 32960、808 或 MQTT 数据来源。", SourceCount: sourceCount, OnlineSourceCount: onlineSourceCount, } } if onlineSourceCount == 0 { return &VehicleServiceStatus{ Status: "offline", Severity: "error", Title: "车辆离线", Detail: "所有已知数据来源均未在线,需要检查平台转发、终端上报或链路状态。", SourceCount: sourceCount, OnlineSourceCount: onlineSourceCount, } } if len(missingProtocols) > 0 { return &VehicleServiceStatus{ Status: "degraded", Severity: "warning", Title: "来源不完整", Detail: sourceCoverageDetail(sourceCount, onlineSourceCount, "车辆服务可用但缺少 "+strings.Join(missingProtocols, "、")+" 来源。"), SourceCount: sourceCount, OnlineSourceCount: onlineSourceCount, } } if onlineSourceCount < sourceCount { return &VehicleServiceStatus{ Status: "degraded", Severity: "warning", Title: "来源不完整", Detail: sourceCoverageDetail(sourceCount, onlineSourceCount, "车辆服务可用但需要关注离线来源。"), SourceCount: sourceCount, OnlineSourceCount: onlineSourceCount, } } return &VehicleServiceStatus{ Status: "healthy", Severity: "ok", Title: "服务正常", Detail: sourceCoverageDetail(sourceCount, onlineSourceCount, "全部已知来源在线。"), SourceCount: sourceCount, OnlineSourceCount: onlineSourceCount, } } func missingProtocolsFromStatuses(statuses []VehicleSourceStatus) []string { missing := make([]string, 0, len(statuses)) for _, status := range statuses { if vehicleSourceEvidenceMissing(status) { missing = append(missing, status.Protocol) } } return missing } func buildVehicleCoverageServiceStatus(row VehicleCoverageRow) *VehicleServiceStatus { if row.BindingStatus != "bound" { return &VehicleServiceStatus{ Status: "identity_required", Severity: "warning", Title: "身份未绑定", Detail: "车辆身份绑定不完整,需先维护 VIN、车牌或手机号映射。", SourceCount: row.SourceCount, OnlineSourceCount: row.OnlineSourceCount, } } if row.SourceCount == 0 { return &VehicleServiceStatus{ Status: "no_data", Severity: "warning", Title: "暂无数据来源", Detail: "车辆已绑定,但暂未查询到 32960、808 或 MQTT 数据来源。", SourceCount: row.SourceCount, OnlineSourceCount: row.OnlineSourceCount, } } if row.OnlineSourceCount == 0 { return &VehicleServiceStatus{ Status: "offline", Severity: "error", Title: "车辆离线", Detail: "所有已知数据来源均未在线,需要检查平台转发、终端上报或链路状态。", SourceCount: row.SourceCount, OnlineSourceCount: row.OnlineSourceCount, } } if len(row.MissingProtocols) > 0 { return &VehicleServiceStatus{ Status: "degraded", Severity: "warning", Title: "来源不完整", Detail: sourceCoverageDetail(row.SourceCount, row.OnlineSourceCount, "车辆服务可用但缺少 "+strings.Join(row.MissingProtocols, "、")+" 来源。"), SourceCount: row.SourceCount, OnlineSourceCount: row.OnlineSourceCount, } } if row.OnlineSourceCount < row.SourceCount { return &VehicleServiceStatus{ Status: "degraded", Severity: "warning", Title: "来源不完整", Detail: sourceCoverageDetail(row.SourceCount, row.OnlineSourceCount, "车辆服务可用但需要关注离线来源。"), SourceCount: row.SourceCount, OnlineSourceCount: row.OnlineSourceCount, } } return &VehicleServiceStatus{ Status: "healthy", Severity: "ok", Title: "服务正常", Detail: sourceCoverageDetail(row.SourceCount, row.OnlineSourceCount, "全部来源在线。"), SourceCount: row.SourceCount, OnlineSourceCount: row.OnlineSourceCount, } } func buildVehicleCoverageSourceConsistency(row VehicleCoverageRow) *VehicleSourceConsistency { consistency := &VehicleSourceConsistency{ SourceCount: row.SourceCount, OnlineSourceCount: row.OnlineSourceCount, MissingProtocols: row.MissingProtocols, Scope: "coverage", LocatedSourceCount: 0, } applyVehicleSourceConsistencyDiagnosis(consistency) return consistency } func buildRealtimeServiceStatus(row VehicleRealtimeRow) *VehicleServiceStatus { return buildVehicleCoverageServiceStatus(VehicleCoverageRow{ VIN: row.VIN, Plate: row.Plate, Phone: row.Phone, OEM: row.OEM, Protocols: row.Protocols, MissingProtocols: missingCanonicalProtocols(row.Protocols), SourceCount: row.SourceCount, OnlineSourceCount: row.OnlineSourceCount, Online: row.Online, LastSeen: row.LastSeen, BindingStatus: row.BindingStatus, }) } func sourceCoverageDetail(sourceCount int, onlineSourceCount int, suffix string) string { return strconv.Itoa(onlineSourceCount) + "/" + strconv.Itoa(sourceCount) + " 个来源在线," + suffix } func latestString(left string, right string) string { left = strings.TrimSpace(left) right = strings.TrimSpace(right) if left == "" { return right } if right == "" || left >= right { return left } return right } func sourceNames(statuses []VehicleSourceStatus) []string { out := make([]string, 0, len(statuses)) for _, status := range statuses { out = append(out, status.Protocol) } return out } func appendIfMissing(values []string, value string) []string { value = strings.TrimSpace(value) if value == "" { return values } for _, current := range values { if current == value { return values } } return append(values, value) } func buildVehicleIdentityResolution(keyword string, vehicles []VehicleRow) VehicleIdentityResolution { keyword = strings.TrimSpace(keyword) result := VehicleIdentityResolution{LookupKey: keyword, Protocols: []string{}} identity := resolveVehicleIdentity(keyword, vehicles) if identity == nil || strings.TrimSpace(identity.VIN) == "" { if isLikelyVIN(keyword) { result.Resolved = true result.VIN = keyword } return result } result.Resolved = true result.VIN = identity.VIN result.Plate = identity.Plate result.Phone = identity.Phone result.OEM = identity.OEM for _, vehicle := range vehicles { if !strings.EqualFold(vehicle.VIN, identity.VIN) { continue } if result.ServiceStatus == nil && vehicle.ServiceStatus != nil { result.ServiceStatus = vehicle.ServiceStatus } if vehicle.Online { result.Online = true } if vehicle.LastSeen > result.LastSeen { result.LastSeen = vehicle.LastSeen } if result.Plate == "" { result.Plate = vehicle.Plate } if result.Phone == "" { result.Phone = vehicle.Phone } if result.OEM == "" { result.OEM = vehicle.OEM } result.Protocols = appendIfMissing(result.Protocols, vehicle.Protocol) } sort.Strings(result.Protocols) if result.ServiceStatus == nil { result.ServiceStatus = buildVehicleCoverageServiceStatus(VehicleCoverageRow{ VIN: result.VIN, Plate: result.Plate, Phone: result.Phone, OEM: result.OEM, Protocols: result.Protocols, SourceCount: len(result.Protocols), OnlineSourceCount: boolToInt(result.Online), Online: result.Online, LastSeen: result.LastSeen, BindingStatus: "bound", }) } return result } func boolToInt(value bool) int { if value { return 1 } return 0 }