shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(platform): surface no-source quality issues

Browse Source
This commit is contained in:
lingniu
2026-07-04 04:34:32 +08:00
parent 40b55600c7
commit 669390e8bb
3 changed files with 115 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
vehicle-data-platform/apps/api/internal/platform/handler_test.go
View File

@@ -5,6 +5,7 @@ import (
"encoding/json"
"net/http"
"net/http/httptest"
"net/url"
"strings"
"testing"
)
@@ -586,6 +587,66 @@ func TestHandlerHistoryMileageQualityOps(t *testing.T) {
}
}
func TestHandlerQualityIncludesNoSourceVehicles(t *testing.T) {
handler := NewHandler(NewService(NewMockStore()))
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/api/quality/issues?limit=20", 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 Page[QualityIssueRow] `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())
}
found := false
for _, item := range body.Data.Items {
if item.IssueType == "NO_SOURCE" {
found = true
if item.VIN == "" || item.Plate == "" || item.Protocol != "VEHICLE_SERVICE" {
t.Fatalf("NO_SOURCE issue should identify bound vehicle service, got %+v", item)
}
}
}
if !found {
t.Fatalf("quality issues should include NO_SOURCE vehicles, got %+v", body.Data.Items)
}
}
func TestHandlerQualitySummaryIncludesNoSourceBucket(t *testing.T) {
handler := NewHandler(NewService(NewMockStore()))
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/api/quality/summary?limit=20", 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 QualitySummary `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())
}
found := false
for _, item := range body.Data.IssueTypes {
if item.Name == "NO_SOURCE" {
found = true
}
}
if !found {
t.Fatalf("quality summary should include NO_SOURCE bucket, got %+v", body.Data.IssueTypes)
}
}
func TestBuildQualityIssueWhereUsesMatchingArgs(t *testing.T) {
fromSQL, args := buildQualityIssueWhere(url.Values{"vin": {"VIN001"}, "protocol": {"VEHICLE_SERVICE"}, "keyword": {"粤A"}})
if strings.Count(fromSQL, "?") != len(args) {
t.Fatalf("quality issue SQL placeholders should match args: placeholders=%d args=%d sql=%s args=%#v", strings.Count(fromSQL, "?"), len(args), fromSQL, args)
}
}
func TestHandlerOpsHealthIncludesVehicleServiceRuntime(t *testing.T) {
handler := NewHandler(NewServiceWithRuntime(NewMockStore(), RuntimeInfo{RequestTimeoutMs: 1500}))
rec := httptest.NewRecorder()

1
vehicle-data-platform/apps/api/internal/platform/mock_store.go
View File

@@ -556,6 +556,7 @@ func mockQualityRows() []QualityIssueRow {
return []QualityIssueRow{
{VIN: "unknown", Plate: "", Phone: "13300000001", SourceEndpoint: "127.0.0.1:808", Protocol: "JT808", IssueType: "VIN_MISSING", Severity: "warning", LastSeen: "2026-07-03 19:58:00", Detail: "phone 13300000001 未命中 binding 表"},
{VIN: "LNXNEGRR7SR318212", Plate: "川AHTWO1", Phone: "", SourceEndpoint: "Hyundai", Protocol: "GB32960", IssueType: "LINK_GAP", Severity: "error", LastSeen: "2026-07-03 18:20:00", Detail: "Hyundai 平台近 60 分钟无转发"},
{VIN: "LB9NO_SOURCE001", Plate: "粤A无源1", Phone: "13300000002", SourceEndpoint: "vehicle_identity_binding", Protocol: "VEHICLE_SERVICE", IssueType: "NO_SOURCE", Severity: "warning", LastSeen: "", Detail: "车辆已绑定，但暂无 32960、808 或 MQTT 数据来源"},
}
}

79
vehicle-data-platform/apps/api/internal/platform/production_store.go
View File

@@ -4,7 +4,6 @@ import (
"context"
"database/sql"
"encoding/json"
"fmt"
"net/url"
"strconv"
"strings"
@@ -704,36 +703,74 @@ func (s *ProductionStore) DailyMileage(ctx context.Context, query url.Values) (P
}
func buildQualityIssueWhere(query url.Values) (string, []any) {
where := []string{"(r.vin = '' OR r.vin = 'unknown' OR r.vin IS NULL)"}
where := []string{"1 = 1"}
args := []any{}
if vin := strings.TrimSpace(query.Get("vin")); vin != "" {
where = append(where, "(r.vin = ? OR b.vin = ?)")
args = append(args, vin, vin)
where = append(where, "q.vin = ?")
args = append(args, vin)
}
if protocol := strings.TrimSpace(query.Get("protocol")); protocol != "" {
where = append(where, "? = 'JT808'")
where = append(where, "q.protocol = ?")
args = append(args, protocol)
}
if keyword := strings.TrimSpace(query.Get("keyword")); keyword != "" {
where = append(where, "(r.phone LIKE ? OR COALESCE(r.source_endpoint, '') LIKE ? OR COALESCE(b.plate, '') LIKE ?)")
where = append(where, "(q.vin LIKE ? OR q.plate LIKE ? OR q.phone LIKE ? OR q.source_endpoint LIKE ?)")
like := "%" + keyword + "%"
args = append(args, like, like, like)
args = append(args, like, like, like, like)
}
return `FROM jt808_registration r LEFT JOIN vehicle_identity_binding b ON b.phone = r.phone WHERE ` + strings.Join(where, " AND "), args
sourceSQL := `SELECT COALESCE(NULLIF(b.vin, ''), 'unknown') AS vin, COALESCE(b.plate, '') AS plate, r.phone AS phone, ` +
`COALESCE(r.source_endpoint, '') AS source_endpoint, 'JT808' AS protocol, 'VIN_MISSING' AS issue_type, ` +
`'warning' AS severity, COALESCE(DATE_FORMAT(r.latest_seen_at, '%Y-%m-%d %H:%i:%s'), '') AS last_seen, ` +
`CONCAT('phone ', r.phone, ' 未命中 binding 表，来源 ', COALESCE(r.source_endpoint, '')) AS detail ` +
`FROM jt808_registration r LEFT JOIN vehicle_identity_binding b ON b.phone = r.phone ` +
`WHERE r.vin = '' OR r.vin = 'unknown' OR r.vin IS NULL ` +
`UNION ALL ` +
`SELECT b.vin AS vin, COALESCE(b.plate, '') AS plate, COALESCE(b.phone, '') AS phone, ` +
`'vehicle_identity_binding' AS source_endpoint, 'VEHICLE_SERVICE' AS protocol, 'NO_SOURCE' AS issue_type, ` +
`'warning' AS severity, '' AS last_seen, '车辆已绑定，但暂无 32960、808 或 MQTT 数据来源' AS detail ` +
`FROM vehicle_identity_binding b LEFT JOIN vehicle_realtime_snapshot s ON s.vin = b.vin ` +
`WHERE b.vin IS NOT NULL AND b.vin <> '' ` +
`GROUP BY b.vin, b.plate, b.phone HAVING COUNT(DISTINCT s.protocol) = 0`
return `FROM (` + sourceSQL + `) q WHERE ` + strings.Join(where, " AND "), args
}
func (s *ProductionStore) QualitySummary(ctx context.Context, query url.Values) (QualitySummary, error) {
fromSQL, args := buildQualityIssueWhere(query)
var summary QualitySummary
if err := s.db.QueryRowContext(ctx, `SELECT COUNT(DISTINCT COALESCE(NULLIF(b.vin, ''), r.phone)) AS issue_vehicle_count, COUNT(*) AS issue_record_count `+fromSQL, args...).Scan(&summary.IssueVehicleCount, &summary.IssueRecordCount); err != nil {
if err := s.db.QueryRowContext(ctx, `SELECT COUNT(DISTINCT COALESCE(NULLIF(q.vin, ''), q.phone)) AS issue_vehicle_count, COUNT(*) AS issue_record_count `+fromSQL, args...).Scan(&summary.IssueVehicleCount, &summary.IssueRecordCount); err != nil {
return QualitySummary{}, err
}
summary.WarningCount = summary.IssueRecordCount
summary.Protocols = []QualityBucketStat{{Name: "JT808", Count: summary.IssueRecordCount}}
summary.IssueTypes = []QualityBucketStat{{Name: "VIN_MISSING", Count: summary.IssueRecordCount}}
protocols, err := s.qualityBuckets(ctx, fromSQL, args, "q.protocol")
if err != nil {
return QualitySummary{}, err
}
issueTypes, err := s.qualityBuckets(ctx, fromSQL, args, "q.issue_type")
if err != nil {
return QualitySummary{}, err
}
summary.Protocols = protocols
summary.IssueTypes = issueTypes
return summary, nil
}
func (s *ProductionStore) qualityBuckets(ctx context.Context, fromSQL string, args []any, field string) ([]QualityBucketStat, error) {
rows, err := s.db.QueryContext(ctx, `SELECT `+field+`, COUNT(*) `+fromSQL+` GROUP BY `+field+` ORDER BY COUNT(*) DESC, `+field+` ASC`, args...)
if err != nil {
return nil, err
}
defer rows.Close()
out := []QualityBucketStat{}
for rows.Next() {
var item QualityBucketStat
if err := rows.Scan(&item.Name, &item.Count); err != nil {
return nil, err
}
out = append(out, item)
}
return out, rows.Err()
}
func (s *ProductionStore) QualityIssues(ctx context.Context, query url.Values) (Page[QualityIssueRow], error) {
limit, offset := buildLimitOffset(query)
fromSQL, args := buildQualityIssueWhere(query)
@@ -743,29 +780,19 @@ func (s *ProductionStore) QualityIssues(ctx context.Context, query url.Values) (
if err := s.db.QueryRowContext(ctx, `SELECT COUNT(*) `+fromSQL, countArgs...).Scan(&total); err != nil {
return Page[QualityIssueRow]{}, err
}
rows, err := s.db.QueryContext(ctx, `SELECT COALESCE(b.vin, 'unknown') AS vin, COALESCE(b.plate, '') AS plate, r.phone, COALESCE(r.source_endpoint, ''), COALESCE(DATE_FORMAT(r.latest_seen_at, '%Y-%m-%d %H:%i:%s'), '') `+
fromSQL+` ORDER BY r.latest_seen_at DESC, r.phone ASC LIMIT ? OFFSET ?`, args...)
rows, err := s.db.QueryContext(ctx, `SELECT q.vin, q.plate, q.phone, q.source_endpoint, q.protocol, q.issue_type, q.severity, q.last_seen, q.detail `+
fromSQL+` ORDER BY q.last_seen DESC, q.vin ASC, q.phone ASC LIMIT ? OFFSET ?`, args...)
if err != nil {
return Page[QualityIssueRow]{}, err
}
defer rows.Close()
items := make([]QualityIssueRow, 0)
for rows.Next() {
var vin, plate, phone, source, lastSeen string
if err := rows.Scan(&vin, &plate, &phone, &source, &lastSeen); err != nil {
var item QualityIssueRow
if err := rows.Scan(&item.VIN, &item.Plate, &item.Phone, &item.SourceEndpoint, &item.Protocol, &item.IssueType, &item.Severity, &item.LastSeen, &item.Detail); err != nil {
return Page[QualityIssueRow]{}, err
}
items = append(items, QualityIssueRow{
VIN: vin,
Plate: plate,
Phone: phone,
SourceEndpoint: source,
Protocol: "JT808",
IssueType: "VIN_MISSING",
Severity: "warning",
LastSeen: lastSeen,
Detail: fmt.Sprintf("phone %s 未命中 binding 表，来源 %s", phone, source),
})
items = append(items, item)
}
return Page[QualityIssueRow]{Items: items, Total: total, Limit: limit, Offset: offset}, rows.Err()
}