307 lines
9.2 KiB
Go
307 lines
9.2 KiB
Go
package stats
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"errors"
|
|
"fmt"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
|
|
)
|
|
|
|
type Execer interface {
|
|
ExecContext(context.Context, string, ...any) (sql.Result, error)
|
|
}
|
|
|
|
type Writer struct {
|
|
exec Execer
|
|
loc *time.Location
|
|
mu sync.Mutex
|
|
lastTotalMileage map[string]float64
|
|
}
|
|
|
|
type MetricSample struct {
|
|
VIN string
|
|
Protocol envelope.Protocol
|
|
StatDate string
|
|
TotalMileageKM float64
|
|
EventTime time.Time
|
|
SourceKey string
|
|
Phone string
|
|
DeviceID string
|
|
SourceEndpoint string
|
|
}
|
|
|
|
func NewWriter(exec Execer, loc *time.Location) *Writer {
|
|
if exec == nil {
|
|
panic("stats execer must not be nil")
|
|
}
|
|
if loc == nil {
|
|
loc = time.FixedZone("Asia/Shanghai", 8*3600)
|
|
}
|
|
return &Writer{exec: exec, loc: loc, lastTotalMileage: map[string]float64{}}
|
|
}
|
|
|
|
func (w *Writer) EnsureSchema(ctx context.Context) error {
|
|
if _, err := w.exec.ExecContext(ctx, DailyMileageTableSQL); err != nil {
|
|
return err
|
|
}
|
|
for _, statement := range DailyMileageAlterSQL {
|
|
if _, err := w.exec.ExecContext(ctx, statement); err != nil && !isDuplicateColumnError(err) {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error {
|
|
samples, err := SamplesFromEnvelope(env, w.loc)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, sample := range samples {
|
|
if w.seenSameMileage(sample) {
|
|
continue
|
|
}
|
|
if _, err := w.exec.ExecContext(ctx, upsertDailyMileageSQL,
|
|
sample.VIN,
|
|
sample.StatDate,
|
|
string(sample.Protocol),
|
|
sample.TotalMileageKM,
|
|
sample.TotalMileageKM,
|
|
sample.SourceKey,
|
|
sample.Phone,
|
|
sample.SourceEndpoint); err != nil {
|
|
return err
|
|
}
|
|
w.markMileageWritten(sample)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (w *Writer) seenSameMileage(sample MetricSample) bool {
|
|
prefix := fmt.Sprintf("%s|%s|", sample.VIN, sample.Protocol)
|
|
key := prefix + sample.StatDate
|
|
w.mu.Lock()
|
|
defer w.mu.Unlock()
|
|
if last, ok := w.lastTotalMileage[key]; ok && last == sample.TotalMileageKM {
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (w *Writer) markMileageWritten(sample MetricSample) {
|
|
prefix := fmt.Sprintf("%s|%s|", sample.VIN, sample.Protocol)
|
|
key := prefix + sample.StatDate
|
|
w.mu.Lock()
|
|
defer w.mu.Unlock()
|
|
for existing := range w.lastTotalMileage {
|
|
if strings.HasPrefix(existing, prefix) && existing != key {
|
|
delete(w.lastTotalMileage, existing)
|
|
}
|
|
}
|
|
w.lastTotalMileage[key] = sample.TotalMileageKM
|
|
}
|
|
|
|
func SamplesFromEnvelope(env envelope.FrameEnvelope, loc *time.Location) ([]MetricSample, error) {
|
|
vin := strings.TrimSpace(env.VIN)
|
|
if vin == "" {
|
|
return nil, nil
|
|
}
|
|
totalMileage, ok := totalMileageKMFromEnvelope(env)
|
|
if !ok {
|
|
return nil, nil
|
|
}
|
|
if totalMileage <= 0 {
|
|
return nil, nil
|
|
}
|
|
if loc == nil {
|
|
loc = time.FixedZone("Asia/Shanghai", 8*3600)
|
|
}
|
|
eventMS := env.EventTimeMS
|
|
if eventMS <= 0 {
|
|
eventMS = env.ReceivedAtMS
|
|
}
|
|
if eventMS <= 0 {
|
|
return nil, errors.New("event or received time is required")
|
|
}
|
|
statDate := time.UnixMilli(eventMS).In(loc).Format("2006-01-02")
|
|
return []MetricSample{{
|
|
VIN: vin,
|
|
Protocol: env.Protocol,
|
|
StatDate: statDate,
|
|
TotalMileageKM: totalMileage,
|
|
EventTime: time.UnixMilli(eventMS).In(loc),
|
|
SourceKey: sourceKey(env),
|
|
Phone: strings.TrimSpace(env.Phone),
|
|
DeviceID: strings.TrimSpace(env.DeviceID),
|
|
SourceEndpoint: strings.TrimSpace(env.SourceEndpoint),
|
|
}}, nil
|
|
}
|
|
|
|
func sourceKey(env envelope.FrameEnvelope) string {
|
|
return SourceKey(env.Protocol, env.Phone, env.DeviceID, NormalizeSourceIP(env.SourceEndpoint))
|
|
}
|
|
|
|
func isDuplicateColumnError(err error) bool {
|
|
if err == nil {
|
|
return false
|
|
}
|
|
text := strings.ToLower(err.Error())
|
|
return strings.Contains(text, "duplicate column") || strings.Contains(text, "1060")
|
|
}
|
|
|
|
type mileageFieldMapping struct {
|
|
key string
|
|
scale float64
|
|
}
|
|
|
|
func totalMileageKMFromEnvelope(env envelope.FrameEnvelope) (float64, bool) {
|
|
if value, ok := floatField(env, envelope.FieldTotalMileageKM); ok {
|
|
return value, true
|
|
}
|
|
for _, mapping := range mileageMappingsByProtocol(env.Protocol) {
|
|
value, ok := floatField(env, mapping.key)
|
|
if !ok {
|
|
continue
|
|
}
|
|
return value * mapping.scale, true
|
|
}
|
|
return 0, false
|
|
}
|
|
|
|
func mileageMappingsByProtocol(protocol envelope.Protocol) []mileageFieldMapping {
|
|
switch protocol {
|
|
case envelope.ProtocolGB32960:
|
|
return []mileageFieldMapping{{key: "gb32960.vehicle.total_mileage_km", scale: 1}}
|
|
case envelope.ProtocolJT808:
|
|
return []mileageFieldMapping{{key: "jt808.location.total_mileage_km", scale: 1}}
|
|
case envelope.ProtocolYutongMQTT:
|
|
return []mileageFieldMapping{
|
|
{key: "yutong_mqtt.data.total_mileage", scale: 0.001},
|
|
{key: "yutong_mqtt.root.data.total_mileage", scale: 0.001},
|
|
}
|
|
default:
|
|
return nil
|
|
}
|
|
}
|
|
|
|
const upsertDailyMileageSQL = `
|
|
INSERT INTO vehicle_daily_mileage
|
|
(vin, stat_date, protocol, daily_mileage_km,
|
|
first_total_mileage_km, latest_total_mileage_km,
|
|
trusted_source_key, trusted_phone, trusted_source_endpoint, sample_count)
|
|
VALUES (?, ?, ?, 0, ?, ?, ?, ?, ?, 1)
|
|
ON DUPLICATE KEY UPDATE
|
|
first_total_mileage_km = CASE
|
|
WHEN trusted_source_key IS NOT NULL
|
|
AND trusted_source_key <> ''
|
|
AND trusted_source_key <> VALUES(trusted_source_key)
|
|
AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50
|
|
THEN first_total_mileage_km
|
|
WHEN first_total_mileage_km IS NULL OR first_total_mileage_km <= 0
|
|
THEN VALUES(first_total_mileage_km)
|
|
ELSE LEAST(first_total_mileage_km, VALUES(first_total_mileage_km))
|
|
END,
|
|
latest_total_mileage_km = CASE
|
|
WHEN trusted_source_key IS NOT NULL
|
|
AND trusted_source_key <> ''
|
|
AND trusted_source_key <> VALUES(trusted_source_key)
|
|
AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50
|
|
THEN latest_total_mileage_km
|
|
WHEN latest_total_mileage_km IS NULL OR latest_total_mileage_km <= 0
|
|
THEN VALUES(latest_total_mileage_km)
|
|
ELSE GREATEST(latest_total_mileage_km, VALUES(latest_total_mileage_km))
|
|
END,
|
|
daily_mileage_km = GREATEST(
|
|
CASE
|
|
WHEN trusted_source_key IS NOT NULL
|
|
AND trusted_source_key <> ''
|
|
AND trusted_source_key <> VALUES(trusted_source_key)
|
|
AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50
|
|
THEN latest_total_mileage_km
|
|
WHEN latest_total_mileage_km IS NULL OR latest_total_mileage_km <= 0
|
|
THEN VALUES(latest_total_mileage_km)
|
|
ELSE latest_total_mileage_km
|
|
END,
|
|
CASE
|
|
WHEN trusted_source_key IS NOT NULL
|
|
AND trusted_source_key <> ''
|
|
AND trusted_source_key <> VALUES(trusted_source_key)
|
|
AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50
|
|
THEN latest_total_mileage_km
|
|
ELSE VALUES(latest_total_mileage_km)
|
|
END
|
|
) - CASE
|
|
WHEN trusted_source_key IS NOT NULL
|
|
AND trusted_source_key <> ''
|
|
AND trusted_source_key <> VALUES(trusted_source_key)
|
|
AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50
|
|
THEN first_total_mileage_km
|
|
WHEN first_total_mileage_km IS NULL OR first_total_mileage_km <= 0
|
|
THEN VALUES(first_total_mileage_km)
|
|
ELSE LEAST(first_total_mileage_km, VALUES(first_total_mileage_km))
|
|
END,
|
|
trusted_source_key = CASE
|
|
WHEN trusted_source_key IS NULL OR trusted_source_key = ''
|
|
THEN VALUES(trusted_source_key)
|
|
WHEN trusted_source_key = VALUES(trusted_source_key)
|
|
THEN trusted_source_key
|
|
WHEN ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) <= 50
|
|
THEN VALUES(trusted_source_key)
|
|
ELSE trusted_source_key
|
|
END,
|
|
trusted_phone = CASE
|
|
WHEN trusted_source_key IS NULL OR trusted_source_key = '' OR trusted_source_key = VALUES(trusted_source_key)
|
|
THEN VALUES(trusted_phone)
|
|
ELSE trusted_phone
|
|
END,
|
|
trusted_source_endpoint = CASE
|
|
WHEN trusted_source_key IS NULL OR trusted_source_key = '' OR trusted_source_key = VALUES(trusted_source_key)
|
|
THEN VALUES(trusted_source_endpoint)
|
|
ELSE trusted_source_endpoint
|
|
END,
|
|
sample_count = CASE
|
|
WHEN trusted_source_key IS NOT NULL
|
|
AND trusted_source_key <> ''
|
|
AND trusted_source_key <> VALUES(trusted_source_key)
|
|
AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50
|
|
THEN sample_count
|
|
ELSE sample_count + 1
|
|
END,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
`
|
|
|
|
func floatField(env envelope.FrameEnvelope, key string) (float64, bool) {
|
|
if env.Fields == nil {
|
|
return 0, false
|
|
}
|
|
value, ok := env.Fields[key]
|
|
if !ok || value == nil {
|
|
return 0, false
|
|
}
|
|
switch typed := value.(type) {
|
|
case float64:
|
|
return typed, true
|
|
case float32:
|
|
return float64(typed), true
|
|
case int:
|
|
return float64(typed), true
|
|
case int64:
|
|
return float64(typed), true
|
|
case uint16:
|
|
return float64(typed), true
|
|
case uint32:
|
|
return float64(typed), true
|
|
case string:
|
|
parsed, err := strconv.ParseFloat(strings.TrimSpace(typed), 64)
|
|
return parsed, err == nil
|
|
default:
|
|
return 0, false
|
|
}
|
|
}
|