feat(platform): add quality notification plan

2026-07-04 13:31:36 +08:00
parent 817590c104
commit ff833bc065
9 changed files with 586 additions and 9 deletions
--- a/vehicle-data-platform/apps/api/internal/platform/handler.go
+++ b/vehicle-data-platform/apps/api/internal/platform/handler.go
@@ -44,6 +44,7 @@ func (h *Handler) routes() {
 	h.mux.HandleFunc("GET /api/mileage/daily", h.handleDailyMileage)
 	h.mux.HandleFunc("GET /api/quality/summary", h.handleQualitySummary)
 	h.mux.HandleFunc("GET /api/quality/issues", h.handleQualityIssues)
+	h.mux.HandleFunc("GET /api/quality/notification-plan", h.handleQualityNotificationPlan)
 	h.mux.HandleFunc("GET /api/ops/health", h.handleOpsHealth)
 }

@@ -179,6 +180,11 @@ func (h *Handler) handleQualitySummary(w http.ResponseWriter, r *http.Request) {
 	h.write(w, r, data, err)
 }

+func (h *Handler) handleQualityNotificationPlan(w http.ResponseWriter, r *http.Request) {
+	data, err := h.service.QualityNotificationPlan(r.Context(), r.URL.Query())
+	h.write(w, r, data, err)
+}
+
 func (h *Handler) handleOpsHealth(w http.ResponseWriter, r *http.Request) {
 	data, err := h.service.OpsHealth(r.Context())
 	h.write(w, r, data, err)
--- a/vehicle-data-platform/apps/api/internal/platform/handler_test.go
+++ b/vehicle-data-platform/apps/api/internal/platform/handler_test.go
@@ -859,6 +859,35 @@ func TestHandlerQualityIssuesFiltersIssueType(t *testing.T) {
 	}
 }

+func TestHandlerQualityNotificationPlan(t *testing.T) {
+	handler := NewHandler(NewService(NewMockStore()))
+	rec := httptest.NewRecorder()
+	req := httptest.NewRequest(http.MethodGet, "/api/quality/notification-plan?limit=10", nil)
+	handler.ServeHTTP(rec, req)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status = %d body=%s", rec.Code, rec.Body.String())
+	}
+	var body struct {
+		Data QualityNotificationPlan `json:"data"`
+	}
+	if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
+		t.Fatalf("response JSON should decode: %v body=%s", err, rec.Body.String())
+	}
+	if body.Data.ActiveRuleCount == 0 || body.Data.P0RuleCount == 0 {
+		t.Fatalf("notification plan should expose active P0 rules, got %+v", body.Data)
+	}
+	if len(body.Data.Rules) == 0 || len(body.Data.Policies) == 0 || len(body.Data.PriorityIssues) == 0 {
+		t.Fatalf("notification plan should include rules, policies and priority issues: %+v", body.Data)
+	}
+	first := body.Data.PriorityIssues[0]
+	if first.Priority != "P0" || first.ActionLabel == "" || first.SLA == "" || first.VehicleLabel == "" {
+		t.Fatalf("priority issue should be enriched for notification: %+v", first)
+	}
+	if !strings.Contains(first.NotificationText, "实时定位") || !strings.Contains(first.NotificationText, "#/history") {
+		t.Fatalf("priority issue should include evidence links, got %q", first.NotificationText)
+	}
+}
+
 func TestHandlerOpsHealthIncludesVehicleServiceRuntime(t *testing.T) {
 	handler := NewHandler(NewServiceWithRuntime(NewMockStore(), RuntimeInfo{RequestTimeoutMs: 1500}))
 	rec := httptest.NewRecorder()
--- a/vehicle-data-platform/apps/api/internal/platform/model.go
+++ b/vehicle-data-platform/apps/api/internal/platform/model.go
@@ -291,6 +291,47 @@ type QualityBucketStat struct {
 	Count int    `json:"count"`
 }

+type QualityNotificationPlan struct {
+	Summary         QualitySummary              `json:"summary"`
+	Rules           []QualityAlertRule          `json:"rules"`
+	Policies        []QualityNotificationPolicy `json:"policies"`
+	PriorityIssues  []QualityPriorityIssue      `json:"priorityIssues"`
+	ActiveRuleCount int                         `json:"activeRuleCount"`
+	P0RuleCount     int                         `json:"p0RuleCount"`
+}
+
+type QualityAlertRule struct {
+	IssueType string `json:"issueType"`
+	Title     string `json:"title"`
+	Level     string `json:"level"`
+	Owner     string `json:"owner"`
+	Trigger   string `json:"trigger"`
+	Notify    string `json:"notify"`
+	SLA       string `json:"sla"`
+	Count     int    `json:"count"`
+}
+
+type QualityNotificationPolicy struct {
+	Name      string `json:"name"`
+	Target    string `json:"target"`
+	Channel   string `json:"channel"`
+	Condition string `json:"condition"`
+}
+
+type QualityPriorityIssue struct {
+	QualityIssueRow
+	Priority         string `json:"priority"`
+	ActionLabel      string `json:"actionLabel"`
+	ActionDetail     string `json:"actionDetail"`
+	SLA              string `json:"sla"`
+	VehicleLabel     string `json:"vehicleLabel"`
+	RealtimeHash     string `json:"realtimeHash"`
+	HistoryHash      string `json:"historyHash"`
+	RawHash          string `json:"rawHash"`
+	VehicleHash      string `json:"vehicleHash"`
+	NotificationText string `json:"notificationText"`
+}
+
 type OpsHealth struct {
 	LinkHealth        []LinkHealth `json:"linkHealth"`
 	KafkaLag          *int         `json:"kafkaLag"`
--- a/vehicle-data-platform/apps/api/internal/platform/service.go
+++ b/vehicle-data-platform/apps/api/internal/platform/service.go
@@ -76,6 +76,58 @@ func asClientError(err error) (clientError, bool) {

 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:    "通知解析负责人核对字段映射和 RAW 样本",
+		SLA:       "当日闭环",
+	},
+}
+
+var qualityNotificationPolicies = []QualityNotificationPolicy{
+	{Name: "P0 实时中断", Target: "接入运维 + 业务责任人", Channel: "站内告警 / 邮件 / 企业微信", Condition: "无来源、VIN 缺失、存储不可写"},
+	{Name: "P1 数据质量", Target: "协议解析 + 数据治理", Channel: "站内告警 / 每日汇总邮件", Condition: "字段缺失、链路间断、容量风险"},
+	{Name: "P2 趋势关注", Target: "数智中心", Channel: "周报 / 趋势看板", Condition: "单源车辆、档案缺项、来源覆盖下降"},
+}
+
+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 {
@@ -806,6 +858,46 @@ func (s *Service) QualityIssues(ctx context.Context, query url.Values) (Page[Qua
 	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 {
@@ -815,6 +907,219 @@ func (s *Service) OpsHealth(ctx context.Context) (OpsHealth, error) {
 	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", 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,
+		"RAW证据：" + 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)
 	vin := firstNonEmpty(resolved.Get("vin"), resolved.Get("keyword"))