feat(platform): add vehicle data management console
This commit is contained in:
lingniu
128
vehicle-data-platform/apps/api/internal/platform/handler.go
Normal file
128
vehicle-data-platform/apps/api/internal/platform/handler.go
Normal file
@@ -0,0 +1,128 @@
|
||||
package platform
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"lingniu/vehicle-data-platform/apps/api/internal/httpx"
|
||||
)
|
||||
|
||||
type Handler struct {
|
||||
service *Service
|
||||
mux *http.ServeMux
|
||||
}
|
||||
|
||||
func NewHandler(service *Service) *Handler {
|
||||
h := &Handler{service: service, mux: http.NewServeMux()}
|
||||
h.routes()
|
||||
return h
|
||||
}
|
||||
|
||||
func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
h.mux.ServeHTTP(w, r)
|
||||
}
|
||||
|
||||
func (h *Handler) routes() {
|
||||
h.mux.HandleFunc("GET /api/dashboard/summary", h.handleDashboardSummary)
|
||||
h.mux.HandleFunc("GET /api/vehicles", h.handleVehicles)
|
||||
h.mux.HandleFunc("GET /api/realtime/locations", h.handleRealtimeLocations)
|
||||
h.mux.HandleFunc("GET /api/history/locations", h.handleHistoryLocations)
|
||||
h.mux.HandleFunc("GET /api/history/raw-frames", h.handleRawFramesGet)
|
||||
h.mux.HandleFunc("POST /api/history/raw-frames/query", h.handleRawFramesPost)
|
||||
h.mux.HandleFunc("GET /api/mileage/daily", h.handleDailyMileage)
|
||||
h.mux.HandleFunc("GET /api/quality/issues", h.handleQualityIssues)
|
||||
h.mux.HandleFunc("GET /api/ops/health", h.handleOpsHealth)
|
||||
}
|
||||
|
||||
func (h *Handler) handleDashboardSummary(w http.ResponseWriter, r *http.Request) {
|
||||
data, err := h.service.DashboardSummary(r.Context())
|
||||
h.write(w, r, data, err)
|
||||
}
|
||||
|
||||
func (h *Handler) handleVehicles(w http.ResponseWriter, r *http.Request) {
|
||||
data, err := h.service.Vehicles(r.Context(), r.URL.Query())
|
||||
h.write(w, r, data, err)
|
||||
}
|
||||
|
||||
func (h *Handler) handleRealtimeLocations(w http.ResponseWriter, r *http.Request) {
|
||||
data, err := h.service.RealtimeLocations(r.Context(), r.URL.Query())
|
||||
h.write(w, r, data, err)
|
||||
}
|
||||
|
||||
func (h *Handler) handleHistoryLocations(w http.ResponseWriter, r *http.Request) {
|
||||
data, err := h.service.HistoryLocations(r.Context(), r.URL.Query())
|
||||
h.write(w, r, data, err)
|
||||
}
|
||||
|
||||
func (h *Handler) handleRawFramesGet(w http.ResponseWriter, r *http.Request) {
|
||||
q := r.URL.Query()
|
||||
query := RawFrameQuery{
|
||||
Protocol: q.Get("protocol"),
|
||||
VIN: q.Get("vin"),
|
||||
DateFrom: q.Get("dateFrom"),
|
||||
DateTo: q.Get("dateTo"),
|
||||
Fields: splitCSV(q.Get("fields")),
|
||||
IncludeFields: q.Get("includeFields") == "true",
|
||||
Limit: parsePositive(q.Get("limit"), 100),
|
||||
Offset: parsePositive(q.Get("offset"), 0),
|
||||
}
|
||||
data, err := h.service.RawFrames(r.Context(), query)
|
||||
h.write(w, r, data, err)
|
||||
}
|
||||
|
||||
func (h *Handler) handleRawFramesPost(w http.ResponseWriter, r *http.Request) {
|
||||
defer r.Body.Close()
|
||||
var query RawFrameQuery
|
||||
if err := json.NewDecoder(r.Body).Decode(&query); err != nil {
|
||||
httpx.WriteError(w, http.StatusBadRequest, "BAD_JSON", "请求 JSON 解析失败", err.Error(), traceID(r))
|
||||
return
|
||||
}
|
||||
data, err := h.service.RawFrames(r.Context(), query)
|
||||
h.write(w, r, data, err)
|
||||
}
|
||||
|
||||
func (h *Handler) handleDailyMileage(w http.ResponseWriter, r *http.Request) {
|
||||
data, err := h.service.DailyMileage(r.Context(), r.URL.Query())
|
||||
h.write(w, r, data, err)
|
||||
}
|
||||
|
||||
func (h *Handler) handleQualityIssues(w http.ResponseWriter, r *http.Request) {
|
||||
data, err := h.service.QualityIssues(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)
|
||||
}
|
||||
|
||||
func (h *Handler) write(w http.ResponseWriter, r *http.Request, data any, err error) {
|
||||
if err != nil {
|
||||
httpx.WriteError(w, http.StatusInternalServerError, "INTERNAL", "服务处理失败", err.Error(), traceID(r))
|
||||
return
|
||||
}
|
||||
httpx.WriteOK(w, traceID(r), data)
|
||||
}
|
||||
|
||||
func traceID(r *http.Request) string {
|
||||
if value := r.Header.Get("X-Trace-Id"); value != "" {
|
||||
return value
|
||||
}
|
||||
return "trace-" + time.Now().Format("20060102150405.000000")
|
||||
}
|
||||
|
||||
func splitCSV(value string) []string {
|
||||
if strings.TrimSpace(value) == "" {
|
||||
return nil
|
||||
}
|
||||
parts := strings.Split(value, ",")
|
||||
out := make([]string, 0, len(parts))
|
||||
for _, part := range parts {
|
||||
if trimmed := strings.TrimSpace(part); trimmed != "" {
|
||||
out = append(out, trimmed)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,74 @@
|
||||
package platform
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestHandlerDashboardSummary(t *testing.T) {
|
||||
handler := NewHandler(NewService(NewMockStore()))
|
||||
rec := httptest.NewRecorder()
|
||||
req := httptest.NewRequest(http.MethodGet, "/api/dashboard/summary", nil)
|
||||
handler.ServeHTTP(rec, req)
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d body=%s", rec.Code, rec.Body.String())
|
||||
}
|
||||
if !strings.Contains(rec.Body.String(), "onlineVehicles") {
|
||||
t.Fatalf("response missing onlineVehicles: %s", rec.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandlerVehicles(t *testing.T) {
|
||||
handler := NewHandler(NewService(NewMockStore()))
|
||||
rec := httptest.NewRecorder()
|
||||
req := httptest.NewRequest(http.MethodGet, "/api/vehicles?limit=10", nil)
|
||||
handler.ServeHTTP(rec, req)
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d body=%s", rec.Code, rec.Body.String())
|
||||
}
|
||||
if !strings.Contains(rec.Body.String(), "LB9A32A24R0LS1426") {
|
||||
t.Fatalf("response missing vehicle: %s", rec.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandlerRealtimeLocations(t *testing.T) {
|
||||
handler := NewHandler(NewService(NewMockStore()))
|
||||
rec := httptest.NewRecorder()
|
||||
req := httptest.NewRequest(http.MethodGet, "/api/realtime/locations?limit=10", nil)
|
||||
handler.ServeHTTP(rec, req)
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d body=%s", rec.Code, rec.Body.String())
|
||||
}
|
||||
if !strings.Contains(rec.Body.String(), "LB9A32A24R0LS1426") {
|
||||
t.Fatalf("response missing vehicle: %s", rec.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandlerHistoryMileageQualityOps(t *testing.T) {
|
||||
cases := []struct {
|
||||
path string
|
||||
want string
|
||||
}{
|
||||
{"/api/history/locations?limit=10", "totalMileageKm"},
|
||||
{"/api/history/raw-frames?protocol=GB32960&vin=LB9A32A24R0LS1426&limit=1&includeFields=true", "rawSizeBytes"},
|
||||
{"/api/mileage/daily?limit=10", "dailyMileageKm"},
|
||||
{"/api/quality/issues?limit=10", "issueType"},
|
||||
{"/api/ops/health", "linkHealth"},
|
||||
}
|
||||
handler := NewHandler(NewService(NewMockStore()))
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.path, func(t *testing.T) {
|
||||
rec := httptest.NewRecorder()
|
||||
req := httptest.NewRequest(http.MethodGet, tc.path, nil)
|
||||
handler.ServeHTTP(rec, req)
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d body=%s", rec.Code, rec.Body.String())
|
||||
}
|
||||
if !strings.Contains(rec.Body.String(), tc.want) {
|
||||
t.Fatalf("response missing %q: %s", tc.want, rec.Body.String())
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
193
vehicle-data-platform/apps/api/internal/platform/mock_store.go
Normal file
193
vehicle-data-platform/apps/api/internal/platform/mock_store.go
Normal file
@@ -0,0 +1,193 @@
|
||||
package platform
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"strings"
|
||||
)
|
||||
|
||||
type MockStore struct {
|
||||
vehicles []VehicleRow
|
||||
locations []RealtimeLocationRow
|
||||
}
|
||||
|
||||
func NewMockStore() *MockStore {
|
||||
vehicles := []VehicleRow{
|
||||
{VIN: "LB9A32A24R0LS1426", Plate: "粤AG18312", Phone: "13307795425", OEM: "G7s", Protocol: "JT808", Online: true, LastSeen: "2026-07-03 20:12:10", LocationText: "广东省广州市", BindingScore: 96},
|
||||
{VIN: "LNXNEGRR7SR318212", Plate: "川AHTWO1", Phone: "", OEM: "Hyundai", Protocol: "GB32960", Online: true, LastSeen: "2026-07-03 20:12:06", LocationText: "四川省成都市", BindingScore: 100},
|
||||
{VIN: "LMRKH9AC2R1004087", Plate: "豫A88888", Phone: "", OEM: "宇通", Protocol: "YUTONG_MQTT", Online: true, LastSeen: "2026-07-03 20:11:59", LocationText: "上海市临港", BindingScore: 92},
|
||||
{VIN: "LB9A32A24P0LS1230", Plate: "粤AFF7936", Phone: "13307795426", OEM: "广安车联", Protocol: "JT808", Online: false, LastSeen: "2026-07-03 19:58:00", LocationText: "广东省佛山市", BindingScore: 88},
|
||||
}
|
||||
return &MockStore{
|
||||
vehicles: vehicles,
|
||||
locations: []RealtimeLocationRow{
|
||||
{VIN: vehicles[0].VIN, Plate: vehicles[0].Plate, Protocol: vehicles[0].Protocol, Longitude: 113.2644, Latitude: 23.1291, SpeedKmh: 42.5, SOCPercent: 76.2, TotalMileageKm: 48798.9, LastSeen: vehicles[0].LastSeen},
|
||||
{VIN: vehicles[1].VIN, Plate: vehicles[1].Plate, Protocol: vehicles[1].Protocol, Longitude: 104.0668, Latitude: 30.5728, SpeedKmh: 18.3, SOCPercent: 64.8, TotalMileageKm: 119925, LastSeen: vehicles[1].LastSeen},
|
||||
{VIN: vehicles[2].VIN, Plate: vehicles[2].Plate, Protocol: vehicles[2].Protocol, Longitude: 121.075044, Latitude: 30.590921, SpeedKmh: 27, SOCPercent: 78.4, TotalMileageKm: 119925, LastSeen: vehicles[2].LastSeen},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func (m *MockStore) DashboardSummary(context.Context) (DashboardSummary, error) {
|
||||
return DashboardSummary{
|
||||
OnlineVehicles: 3,
|
||||
ActiveToday: 4,
|
||||
FrameToday: 1286320,
|
||||
IssueVehicles: 7,
|
||||
KafkaLag: 0,
|
||||
Protocols: []ProtocolStat{
|
||||
{Protocol: "GB32960", Online: 18, Total: 64},
|
||||
{Protocol: "JT808", Online: 73, Total: 318},
|
||||
{Protocol: "YUTONG_MQTT", Online: 1, Total: 8},
|
||||
},
|
||||
LinkHealth: []LinkHealth{
|
||||
{Name: "Redis realtime", Status: "ok", Detail: "在线状态 1 分钟 TTL 正常"},
|
||||
{Name: "TDengine raw_frames", Status: "ok", Detail: "最近 5 分钟有写入"},
|
||||
{Name: "Kafka vehicle.fields", Status: "ok", Detail: "consumer lag 0"},
|
||||
{Name: "MySQL identity", Status: "warning", Detail: "部分 808 phone 未绑定 VIN"},
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (m *MockStore) Vehicles(_ context.Context, query url.Values) (Page[VehicleRow], error) {
|
||||
items := filterVehicles(m.vehicles, query)
|
||||
return page(items, query), nil
|
||||
}
|
||||
|
||||
func (m *MockStore) RealtimeLocations(_ context.Context, query url.Values) (Page[RealtimeLocationRow], error) {
|
||||
items := m.locations
|
||||
if vin := strings.TrimSpace(query.Get("vin")); vin != "" {
|
||||
items = keep(items, func(row RealtimeLocationRow) bool { return row.VIN == vin })
|
||||
}
|
||||
if protocol := strings.TrimSpace(query.Get("protocol")); protocol != "" {
|
||||
items = keep(items, func(row RealtimeLocationRow) bool { return row.Protocol == protocol })
|
||||
}
|
||||
return page(items, query), nil
|
||||
}
|
||||
|
||||
func (m *MockStore) HistoryLocations(ctx context.Context, query url.Values) (Page[HistoryLocationRow], error) {
|
||||
realtime, _ := m.RealtimeLocations(ctx, query)
|
||||
rows := make([]HistoryLocationRow, 0, len(realtime.Items))
|
||||
for _, row := range realtime.Items {
|
||||
rows = append(rows, HistoryLocationRow{
|
||||
VIN: row.VIN, Plate: row.Plate, Protocol: row.Protocol, Longitude: row.Longitude, Latitude: row.Latitude,
|
||||
SpeedKmh: row.SpeedKmh, TotalMileageKm: row.TotalMileageKm, DeviceTime: row.LastSeen, ServerTime: row.LastSeen,
|
||||
})
|
||||
}
|
||||
return Page[HistoryLocationRow]{Items: rows, Total: len(rows), Limit: realtime.Limit, Offset: realtime.Offset}, nil
|
||||
}
|
||||
|
||||
func (m *MockStore) RawFrames(_ context.Context, query RawFrameQuery) (Page[RawFrameRow], error) {
|
||||
fields := map[string]any{
|
||||
"gb32960.vehicle.speed_kmh": 42.5,
|
||||
"gb32960.vehicle.total_mileage_km": 48798.9,
|
||||
"gb32960.fuel_cell.stack_1.avg_voltage_v": 0.79,
|
||||
"jt808.location.additional.total_mileage_km": 48798.9,
|
||||
}
|
||||
if len(query.Fields) > 0 {
|
||||
filtered := map[string]any{}
|
||||
for _, key := range query.Fields {
|
||||
if value, ok := fields[key]; ok {
|
||||
filtered[key] = value
|
||||
}
|
||||
}
|
||||
fields = filtered
|
||||
}
|
||||
if !query.IncludeFields && len(query.Fields) == 0 {
|
||||
fields = nil
|
||||
}
|
||||
protocol := defaultString(query.Protocol, "GB32960")
|
||||
vin := defaultString(query.VIN, "LB9A32A24R0LS1426")
|
||||
rows := []RawFrameRow{
|
||||
{ID: "raw-20260703-001", VIN: vin, Protocol: protocol, FrameType: "realtime", DeviceTime: "2026-07-03 20:12:06", ServerTime: "2026-07-03 20:12:06", RawSizeBytes: 430, ParsedFields: fields},
|
||||
}
|
||||
limit := query.Limit
|
||||
if limit <= 0 {
|
||||
limit = 100
|
||||
}
|
||||
return Page[RawFrameRow]{Items: rows, Total: len(rows), Limit: limit, Offset: query.Offset}, nil
|
||||
}
|
||||
|
||||
func (m *MockStore) DailyMileage(_ context.Context, query url.Values) (Page[DailyMileageRow], error) {
|
||||
rows := []DailyMileageRow{
|
||||
{VIN: "LB9A32A24R0LS1426", Plate: "粤AG18312", Date: "2026-07-03", StartMileageKm: 48710.2, EndMileageKm: 48798.9, DailyMileageKm: 88.7, Source: "JT808"},
|
||||
{VIN: "LNXNEGRR7SR318212", Plate: "川AHTWO1", Date: "2026-07-03", StartMileageKm: 119820.4, EndMileageKm: 119925, DailyMileageKm: 104.6, Source: "GB32960"},
|
||||
}
|
||||
return page(rows, query), nil
|
||||
}
|
||||
|
||||
func (m *MockStore) QualityIssues(_ context.Context, query url.Values) (Page[QualityIssueRow], error) {
|
||||
rows := []QualityIssueRow{
|
||||
{VIN: "LB9A32A24P0LS1230", Plate: "粤AFF7936", Protocol: "JT808", IssueType: "VIN_MISSING", Severity: "warning", LastSeen: "2026-07-03 19:58:00", Detail: "phone 未命中 binding 表"},
|
||||
{VIN: "LNXNEGRR7SR318212", Plate: "川AHTWO1", Protocol: "GB32960", IssueType: "LINK_GAP", Severity: "error", LastSeen: "2026-07-03 18:20:00", Detail: "Hyundai 平台近 60 分钟无转发"},
|
||||
}
|
||||
return page(rows, query), nil
|
||||
}
|
||||
|
||||
func (m *MockStore) OpsHealth(context.Context) (OpsHealth, error) {
|
||||
summary, _ := m.DashboardSummary(context.Background())
|
||||
return OpsHealth{
|
||||
LinkHealth: summary.LinkHealth,
|
||||
KafkaLag: summary.KafkaLag,
|
||||
RedisOnlineKeys: 92,
|
||||
TDengineWritable: true,
|
||||
MySQLWritable: true,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func filterVehicles(rows []VehicleRow, query url.Values) []VehicleRow {
|
||||
keyword := strings.ToLower(strings.TrimSpace(query.Get("keyword")))
|
||||
protocol := strings.TrimSpace(query.Get("protocol"))
|
||||
if keyword == "" && protocol == "" {
|
||||
return rows
|
||||
}
|
||||
return keep(rows, func(row VehicleRow) bool {
|
||||
if protocol != "" && row.Protocol != protocol {
|
||||
return false
|
||||
}
|
||||
if keyword == "" {
|
||||
return true
|
||||
}
|
||||
value := strings.ToLower(row.VIN + row.Plate + row.Phone + row.OEM)
|
||||
return strings.Contains(value, keyword)
|
||||
})
|
||||
}
|
||||
|
||||
func page[T any](rows []T, query url.Values) Page[T] {
|
||||
limit := parsePositive(query.Get("limit"), 20)
|
||||
offset := parsePositive(query.Get("offset"), 0)
|
||||
if offset > len(rows) {
|
||||
offset = len(rows)
|
||||
}
|
||||
end := offset + limit
|
||||
if end > len(rows) {
|
||||
end = len(rows)
|
||||
}
|
||||
return Page[T]{Items: rows[offset:end], Total: len(rows), Limit: limit, Offset: offset}
|
||||
}
|
||||
|
||||
func keep[T any](rows []T, fn func(T) bool) []T {
|
||||
out := make([]T, 0, len(rows))
|
||||
for _, row := range rows {
|
||||
if fn(row) {
|
||||
out = append(out, row)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func parsePositive(raw string, fallback int) int {
|
||||
value, err := strconv.Atoi(raw)
|
||||
if err != nil || value < 0 {
|
||||
return fallback
|
||||
}
|
||||
return value
|
||||
}
|
||||
|
||||
func defaultString(value, fallback string) string {
|
||||
if strings.TrimSpace(value) == "" {
|
||||
return fallback
|
||||
}
|
||||
return value
|
||||
}
|
||||
106
vehicle-data-platform/apps/api/internal/platform/model.go
Normal file
106
vehicle-data-platform/apps/api/internal/platform/model.go
Normal file
@@ -0,0 +1,106 @@
|
||||
package platform
|
||||
|
||||
type Page[T any] struct {
|
||||
Items []T `json:"items"`
|
||||
Total int `json:"total"`
|
||||
Limit int `json:"limit"`
|
||||
Offset int `json:"offset"`
|
||||
}
|
||||
|
||||
type DashboardSummary struct {
|
||||
OnlineVehicles int `json:"onlineVehicles"`
|
||||
ActiveToday int `json:"activeToday"`
|
||||
FrameToday int `json:"frameToday"`
|
||||
IssueVehicles int `json:"issueVehicles"`
|
||||
KafkaLag int `json:"kafkaLag"`
|
||||
Protocols []ProtocolStat `json:"protocols"`
|
||||
LinkHealth []LinkHealth `json:"linkHealth"`
|
||||
}
|
||||
|
||||
type ProtocolStat struct {
|
||||
Protocol string `json:"protocol"`
|
||||
Online int `json:"online"`
|
||||
Total int `json:"total"`
|
||||
}
|
||||
|
||||
type LinkHealth struct {
|
||||
Name string `json:"name"`
|
||||
Status string `json:"status"`
|
||||
Detail string `json:"detail,omitempty"`
|
||||
}
|
||||
|
||||
type VehicleRow struct {
|
||||
VIN string `json:"vin"`
|
||||
Plate string `json:"plate"`
|
||||
Phone string `json:"phone"`
|
||||
OEM string `json:"oem"`
|
||||
Protocol string `json:"protocol"`
|
||||
Online bool `json:"online"`
|
||||
LastSeen string `json:"lastSeen"`
|
||||
LocationText string `json:"locationText"`
|
||||
BindingScore int `json:"bindingScore"`
|
||||
}
|
||||
|
||||
type RealtimeLocationRow struct {
|
||||
VIN string `json:"vin"`
|
||||
Plate string `json:"plate"`
|
||||
Protocol string `json:"protocol"`
|
||||
Longitude float64 `json:"longitude"`
|
||||
Latitude float64 `json:"latitude"`
|
||||
SpeedKmh float64 `json:"speedKmh"`
|
||||
SOCPercent float64 `json:"socPercent"`
|
||||
TotalMileageKm float64 `json:"totalMileageKm"`
|
||||
LastSeen string `json:"lastSeen"`
|
||||
}
|
||||
|
||||
type HistoryLocationRow struct {
|
||||
VIN string `json:"vin"`
|
||||
Plate string `json:"plate"`
|
||||
Protocol string `json:"protocol"`
|
||||
Longitude float64 `json:"longitude"`
|
||||
Latitude float64 `json:"latitude"`
|
||||
SpeedKmh float64 `json:"speedKmh"`
|
||||
TotalMileageKm float64 `json:"totalMileageKm"`
|
||||
DeviceTime string `json:"deviceTime"`
|
||||
ServerTime string `json:"serverTime"`
|
||||
}
|
||||
|
||||
type RawFrameRow struct {
|
||||
ID string `json:"id"`
|
||||
VIN string `json:"vin"`
|
||||
Protocol string `json:"protocol"`
|
||||
FrameType string `json:"frameType"`
|
||||
DeviceTime string `json:"deviceTime"`
|
||||
ServerTime string `json:"serverTime"`
|
||||
RawSizeBytes int `json:"rawSizeBytes"`
|
||||
ParsedFields map[string]any `json:"parsedFields"`
|
||||
}
|
||||
|
||||
type DailyMileageRow struct {
|
||||
VIN string `json:"vin"`
|
||||
Plate string `json:"plate"`
|
||||
Date string `json:"date"`
|
||||
StartMileageKm float64 `json:"startMileageKm"`
|
||||
EndMileageKm float64 `json:"endMileageKm"`
|
||||
DailyMileageKm float64 `json:"dailyMileageKm"`
|
||||
Source string `json:"source"`
|
||||
AnomalySeverity string `json:"anomalySeverity,omitempty"`
|
||||
}
|
||||
|
||||
type QualityIssueRow struct {
|
||||
VIN string `json:"vin"`
|
||||
Plate string `json:"plate"`
|
||||
Protocol string `json:"protocol"`
|
||||
IssueType string `json:"issueType"`
|
||||
Severity string `json:"severity"`
|
||||
LastSeen string `json:"lastSeen"`
|
||||
Detail string `json:"detail"`
|
||||
}
|
||||
|
||||
type OpsHealth struct {
|
||||
LinkHealth []LinkHealth `json:"linkHealth"`
|
||||
KafkaLag int `json:"kafkaLag"`
|
||||
RedisOnlineKeys int `json:"redisOnlineKeys"`
|
||||
TDengineWritable bool `json:"tdengineWritable"`
|
||||
MySQLWritable bool `json:"mysqlWritable"`
|
||||
}
|
||||
@@ -0,0 +1,81 @@
|
||||
package platform
|
||||
|
||||
import (
|
||||
"net/url"
|
||||
"strconv"
|
||||
"strings"
|
||||
)
|
||||
|
||||
type SQLQuery struct {
|
||||
Text string
|
||||
Args []any
|
||||
}
|
||||
|
||||
func buildVehicleListSQL(query url.Values) SQLQuery {
|
||||
limit := parsePositive(query.Get("limit"), 20)
|
||||
offset := parsePositive(query.Get("offset"), 0)
|
||||
args := []any{}
|
||||
where := []string{"1 = 1"}
|
||||
if keyword := strings.TrimSpace(query.Get("keyword")); keyword != "" {
|
||||
where = append(where, "(b.vin LIKE ? OR b.plate LIKE ? OR b.phone LIKE ? OR b.oem LIKE ?)")
|
||||
like := "%" + keyword + "%"
|
||||
args = append(args, like, like, like, like)
|
||||
}
|
||||
if protocol := strings.TrimSpace(query.Get("protocol")); protocol != "" {
|
||||
where = append(where, "s.protocol = ?")
|
||||
args = append(args, protocol)
|
||||
}
|
||||
args = append(args, limit, offset)
|
||||
return SQLQuery{
|
||||
Text: `SELECT b.vin, b.plate, b.phone, b.oem, COALESCE(s.protocol, '') AS protocol, ` +
|
||||
`CASE WHEN s.updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) THEN 1 ELSE 0 END AS online, ` +
|
||||
`COALESCE(DATE_FORMAT(s.updated_at, '%Y-%m-%d %H:%i:%s'), '') AS last_seen, ` +
|
||||
`COALESCE(CONCAT(l.longitude, ',', l.latitude), '') AS location_text, ` +
|
||||
`CASE WHEN b.vin IS NOT NULL AND b.vin <> '' THEN 100 ELSE 0 END AS binding_score ` +
|
||||
`FROM vehicle_identity_binding b ` +
|
||||
`LEFT JOIN vehicle_realtime_snapshot s ON s.vin = b.vin ` +
|
||||
`LEFT JOIN vehicle_realtime_location l ON l.vin = b.vin ` +
|
||||
`WHERE ` + strings.Join(where, " AND ") + ` ORDER BY s.updated_at DESC LIMIT ? OFFSET ?`,
|
||||
Args: args,
|
||||
}
|
||||
}
|
||||
|
||||
func buildDailyMileageSQL(query url.Values) SQLQuery {
|
||||
limit := parsePositive(query.Get("limit"), 20)
|
||||
offset := parsePositive(query.Get("offset"), 0)
|
||||
args := []any{}
|
||||
where := []string{"1 = 1"}
|
||||
if vin := strings.TrimSpace(query.Get("vin")); vin != "" {
|
||||
where = append(where, "m.vin = ?")
|
||||
args = append(args, vin)
|
||||
}
|
||||
if dateFrom := strings.TrimSpace(query.Get("dateFrom")); dateFrom != "" {
|
||||
where = append(where, "m.stat_date >= ?")
|
||||
args = append(args, dateFrom)
|
||||
}
|
||||
if dateTo := strings.TrimSpace(query.Get("dateTo")); dateTo != "" {
|
||||
where = append(where, "m.stat_date <= ?")
|
||||
args = append(args, dateTo)
|
||||
}
|
||||
args = append(args, limit, offset)
|
||||
return SQLQuery{
|
||||
Text: `SELECT m.vin, COALESCE(b.plate, '') AS plate, DATE_FORMAT(m.stat_date, '%Y-%m-%d') AS stat_date, ` +
|
||||
`m.first_total_mileage_km, m.latest_total_mileage_km, m.daily_mileage_km, m.protocol ` +
|
||||
`FROM vehicle_daily_mileage m LEFT JOIN vehicle_identity_binding b ON b.vin = m.vin ` +
|
||||
`WHERE ` + strings.Join(where, " AND ") + ` ORDER BY m.stat_date DESC LIMIT ? OFFSET ?`,
|
||||
Args: args,
|
||||
}
|
||||
}
|
||||
|
||||
func buildLimitOffset(query url.Values) (int, int) {
|
||||
return parsePositive(query.Get("limit"), 20), parsePositive(query.Get("offset"), 0)
|
||||
}
|
||||
|
||||
func parseLimitOffset(rawLimit, rawOffset string) (int, int) {
|
||||
return parsePositive(rawLimit, 20), parsePositive(rawOffset, 0)
|
||||
}
|
||||
|
||||
func mustInt(value string) int {
|
||||
n, _ := strconv.Atoi(value)
|
||||
return n
|
||||
}
|
||||
@@ -0,0 +1,45 @@
|
||||
package platform
|
||||
|
||||
import (
|
||||
"net/url"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestBuildVehicleListSQL(t *testing.T) {
|
||||
query := url.Values{"keyword": {"粤A"}, "protocol": {"JT808"}, "limit": {"5"}, "offset": {"10"}}
|
||||
built := buildVehicleListSQL(query)
|
||||
for _, want := range []string{"vehicle_identity_binding", "vehicle_realtime_snapshot", "vehicle_realtime_location", "ORDER BY s.updated_at DESC"} {
|
||||
if !strings.Contains(built.Text, want) {
|
||||
t.Fatalf("SQL missing %q: %s", want, built.Text)
|
||||
}
|
||||
}
|
||||
if len(built.Args) != 7 || built.Args[0] != "%粤A%" || built.Args[4] != "JT808" || built.Args[5] != 5 || built.Args[6] != 10 {
|
||||
t.Fatalf("args = %#v", built.Args)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildDailyMileageSQL(t *testing.T) {
|
||||
query := url.Values{"vin": {"VIN001"}, "dateFrom": {"2026-07-01"}, "dateTo": {"2026-07-03"}}
|
||||
built := buildDailyMileageSQL(query)
|
||||
if !strings.Contains(built.Text, "vehicle_daily_mileage") || !strings.Contains(built.Text, "vehicle_identity_binding") {
|
||||
t.Fatalf("SQL = %s", built.Text)
|
||||
}
|
||||
if len(built.Args) != 5 || built.Args[0] != "VIN001" || built.Args[3] != 20 || built.Args[4] != 0 {
|
||||
t.Fatalf("args = %#v", built.Args)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildRawFrameSQL(t *testing.T) {
|
||||
built := buildRawFrameSQL("lingniu_vehicle_ts", RawFrameQuery{
|
||||
Protocol: "GB32960",
|
||||
VIN: "VIN001",
|
||||
Limit: 1,
|
||||
})
|
||||
if !strings.Contains(built.Text, "lingniu_vehicle_ts.raw_frames") || !strings.Contains(built.Text, "parsed_fields") {
|
||||
t.Fatalf("SQL = %s", built.Text)
|
||||
}
|
||||
if len(built.Args) != 4 || built.Args[0] != "GB32960" || built.Args[1] != "VIN001" || built.Args[2] != 1 {
|
||||
t.Fatalf("args = %#v", built.Args)
|
||||
}
|
||||
}
|
||||
71
vehicle-data-platform/apps/api/internal/platform/service.go
Normal file
71
vehicle-data-platform/apps/api/internal/platform/service.go
Normal file
@@ -0,0 +1,71 @@
|
||||
package platform
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/url"
|
||||
)
|
||||
|
||||
type Store interface {
|
||||
DashboardSummary(context.Context) (DashboardSummary, error)
|
||||
Vehicles(context.Context, url.Values) (Page[VehicleRow], error)
|
||||
RealtimeLocations(context.Context, url.Values) (Page[RealtimeLocationRow], error)
|
||||
HistoryLocations(context.Context, url.Values) (Page[HistoryLocationRow], error)
|
||||
RawFrames(context.Context, RawFrameQuery) (Page[RawFrameRow], error)
|
||||
DailyMileage(context.Context, url.Values) (Page[DailyMileageRow], error)
|
||||
QualityIssues(context.Context, url.Values) (Page[QualityIssueRow], error)
|
||||
OpsHealth(context.Context) (OpsHealth, error)
|
||||
}
|
||||
|
||||
type RawFrameQuery struct {
|
||||
Protocol string `json:"protocol"`
|
||||
VIN string `json:"vin"`
|
||||
DateFrom string `json:"dateFrom"`
|
||||
DateTo string `json:"dateTo"`
|
||||
Fields []string `json:"fields"`
|
||||
IncludeFields bool `json:"includeFields"`
|
||||
Limit int `json:"limit"`
|
||||
Offset int `json:"offset"`
|
||||
}
|
||||
|
||||
type Service struct {
|
||||
store Store
|
||||
}
|
||||
|
||||
func NewService(store Store) *Service {
|
||||
return &Service{store: store}
|
||||
}
|
||||
|
||||
func (s *Service) DashboardSummary(ctx context.Context) (DashboardSummary, error) {
|
||||
return s.store.DashboardSummary(ctx)
|
||||
}
|
||||
|
||||
func (s *Service) Vehicles(ctx context.Context, query url.Values) (Page[VehicleRow], error) {
|
||||
return s.store.Vehicles(ctx, query)
|
||||
}
|
||||
|
||||
func (s *Service) RealtimeLocations(ctx context.Context, query url.Values) (Page[RealtimeLocationRow], error) {
|
||||
return s.store.RealtimeLocations(ctx, query)
|
||||
}
|
||||
|
||||
func (s *Service) HistoryLocations(ctx context.Context, query url.Values) (Page[HistoryLocationRow], error) {
|
||||
return s.store.HistoryLocations(ctx, query)
|
||||
}
|
||||
|
||||
func (s *Service) RawFrames(ctx context.Context, query RawFrameQuery) (Page[RawFrameRow], error) {
|
||||
if query.Limit <= 0 || query.Limit > 500 {
|
||||
query.Limit = 100
|
||||
}
|
||||
return s.store.RawFrames(ctx, query)
|
||||
}
|
||||
|
||||
func (s *Service) DailyMileage(ctx context.Context, query url.Values) (Page[DailyMileageRow], error) {
|
||||
return s.store.DailyMileage(ctx, query)
|
||||
}
|
||||
|
||||
func (s *Service) QualityIssues(ctx context.Context, query url.Values) (Page[QualityIssueRow], error) {
|
||||
return s.store.QualityIssues(ctx, query)
|
||||
}
|
||||
|
||||
func (s *Service) OpsHealth(ctx context.Context) (OpsHealth, error) {
|
||||
return s.store.OpsHealth(ctx)
|
||||
}
|
||||
@@ -0,0 +1,64 @@
|
||||
package platform
|
||||
|
||||
import "strings"
|
||||
|
||||
func buildRawFrameSQL(database string, query RawFrameQuery) SQLQuery {
|
||||
table := qualifyTDengine(database, "raw_frames")
|
||||
where := []string{"1 = 1"}
|
||||
args := []any{}
|
||||
if query.Protocol != "" {
|
||||
where = append(where, "protocol = ?")
|
||||
args = append(args, query.Protocol)
|
||||
}
|
||||
if query.VIN != "" {
|
||||
where = append(where, "vin = ?")
|
||||
args = append(args, query.VIN)
|
||||
}
|
||||
if query.DateFrom != "" {
|
||||
where = append(where, "ts >= ?")
|
||||
args = append(args, query.DateFrom)
|
||||
}
|
||||
if query.DateTo != "" {
|
||||
where = append(where, "ts <= ?")
|
||||
args = append(args, query.DateTo)
|
||||
}
|
||||
limit := query.Limit
|
||||
if limit <= 0 {
|
||||
limit = 100
|
||||
}
|
||||
args = append(args, limit, query.Offset)
|
||||
return SQLQuery{
|
||||
Text: `SELECT ts, frame_id, event_time, received_at, raw_size_bytes, parsed_fields, parse_status, ` +
|
||||
`parse_error, source_endpoint, protocol, vehicle_key, vin, phone FROM ` + table +
|
||||
` WHERE ` + strings.Join(where, " AND ") + ` ORDER BY ts DESC LIMIT ? OFFSET ?`,
|
||||
Args: args,
|
||||
}
|
||||
}
|
||||
|
||||
func buildHistoryLocationSQL(database string, query map[string]string) SQLQuery {
|
||||
table := qualifyTDengine(database, "vehicle_locations")
|
||||
where := []string{"1 = 1"}
|
||||
args := []any{}
|
||||
if protocol := strings.TrimSpace(query["protocol"]); protocol != "" {
|
||||
where = append(where, "protocol = ?")
|
||||
args = append(args, protocol)
|
||||
}
|
||||
if vin := strings.TrimSpace(query["vin"]); vin != "" {
|
||||
where = append(where, "vin = ?")
|
||||
args = append(args, vin)
|
||||
}
|
||||
limit, offset := parseLimitOffset(query["limit"], query["offset"])
|
||||
args = append(args, limit, offset)
|
||||
return SQLQuery{
|
||||
Text: `SELECT ts, vin, protocol, longitude, latitude, speed_kmh, total_mileage_km, event_time, received_at FROM ` +
|
||||
table + ` WHERE ` + strings.Join(where, " AND ") + ` ORDER BY ts DESC LIMIT ? OFFSET ?`,
|
||||
Args: args,
|
||||
}
|
||||
}
|
||||
|
||||
func qualifyTDengine(database, table string) string {
|
||||
if strings.TrimSpace(database) == "" {
|
||||
return table
|
||||
}
|
||||
return database + "." + table
|
||||
}
|
||||
Reference in New Issue
Block a user