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

fix(mileage): fall through ineffective priority sources

Browse Source
This commit is contained in:
lingniu
2026-07-19 21:28:11 +08:00
parent 5539c5e0c2
commit d0c2bb3316
4 changed files with 36 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
docs/architecture/production-data-plane-inventory.md
View File

@@ -106,6 +106,8 @@ Stat writer 每条有效里程样本都会更新 `vehicle_daily_mileage_source`
每日里程只采用同来源累计里程边界差值：`当日最新累计总里程 - 最近历史日最后累计总里程`。优先取前一自然日；前一日没有数据时继续向更早日期查找，历史完全为空才以当天第一条为基线。不同协议、平台和终端来源分别计算候选，再由来源质量和优先级选举最终结果，不混用两个来源的累计总里程。
车辆数据中台按用户启用的协议顺序查询“单车、单日”结果，但只在有有效增量的来源之间应用优先级：首选协议为 `0 km`、其他已启用协议存在正向里程时，查询层降级到有增量的最高优先级来源；全部来源均为 `0 km` 时仍严格按原优先级返回 `0`。该规则只选择一个完整协议事实，不拼接不同协议的首末累计里程。
`vehicle_data_source` 以 `(protocol, source_ip)` 唯一识别来源。直连终端可能产生很多来源 IP，这些记录可以保留但默认不要求平台名；转发平台来源通过 `vehicle_identifier.source_code` 和 `jt808_registration.source_endpoint` 推断后补充 `source_code/platform_name`，后续统计优先使用 `source_id/trust_priority/enabled` 做可信源选择。
统计分日优先使用协议事件时间；如果事件时间明显晚于接收时间超过 10 分钟，则认为设备时间异常，使用接收时间归属统计日。RAW 历史仍保留原始事件时间，便于追溯。

13
vehicle-data-platform/apps/api/internal/platform/mysql_queries.go
View File

@@ -526,7 +526,7 @@ func buildDailyMileageSQL(query url.Values) SQLQuery {
selectionOrder := `m.daily_mileage_km DESC, m.protocol ASC`
dailyMileageExpression := `MAX(COALESCE(m.daily_mileage_km, 0))`
if len(protocols) > 0 {
selectionOrder = mileageProtocolOrder("m.protocol", protocols)
selectionOrder = mileageDailySelectionOrder("m.daily_mileage_km", "m.protocol", protocols)
dailyMileageExpression = `COALESCE(CAST(SUBSTRING_INDEX(GROUP_CONCAT(CAST(COALESCE(m.daily_mileage_km, 0) AS CHAR) ORDER BY ` + selectionOrder + `), ',', 1) AS DECIMAL(18,3)), 0)`
}
groupSQL := fromSQL + ` GROUP BY m.vin, m.stat_date`
@@ -626,7 +626,7 @@ func buildMileageStatisticsBaseSQL(query url.Values) (string, []any) {
dailyMileageExpression := `MAX(COALESCE(m.daily_mileage_km, 0))`
latestMileageExpression := `MAX(COALESCE(m.latest_total_mileage_km, 0))`
if len(protocols) > 0 {
selectionOrder := mileageProtocolOrder("m.protocol", protocols)
selectionOrder := mileageDailySelectionOrder("m.daily_mileage_km", "m.protocol", protocols)
dailyMileageExpression = `COALESCE(CAST(SUBSTRING_INDEX(GROUP_CONCAT(CAST(COALESCE(m.daily_mileage_km, 0) AS CHAR) ORDER BY ` + selectionOrder + `), ',', 1) AS DECIMAL(18,3)), 0)`
latestMileageExpression = `COALESCE(CAST(SUBSTRING_INDEX(GROUP_CONCAT(CAST(COALESCE(m.latest_total_mileage_km, 0) AS CHAR) ORDER BY ` + selectionOrder + `), ',', 1) AS DECIMAL(18,3)), 0)`
}
@@ -729,6 +729,15 @@ func mileageProtocolOrder(column string, protocols []string) string {
return strings.Join(parts, " ") + ", " + column + " ASC"
}
// mileageDailySelectionOrder keeps the configured protocol priority among
// usable daily facts while preventing a zero-increment source from masking a
// positive result reported by another enabled protocol. If every enabled
// protocol reports zero, the configured priority still decides the source.
func mileageDailySelectionOrder(mileageColumn, protocolColumn string, protocols []string) string {
return "CASE WHEN COALESCE(" + mileageColumn + ", 0) > 0 THEN 0 ELSE 1 END, " +
mileageProtocolOrder(protocolColumn, protocols)
}
func appendVINListFilter(where []string, args []any, column string, raw string) ([]string, []any) {
seen := map[string]bool{}
values := make([]string, 0)

22
vehicle-data-platform/apps/api/internal/platform/query_builders_test.go
View File

@@ -511,6 +511,7 @@ func TestMileageQueriesRespectEnabledProtocolPriority(t *testing.T) {
daily := buildDailyMileageSQL(query)
for _, want := range []string{
"m.protocol IN (?,?)",
"CASE WHEN COALESCE(m.daily_mileage_km, 0) > 0 THEN 0 ELSE 1 END",
"CASE m.protocol WHEN 'JT808' THEN 1 WHEN 'GB32960' THEN 2 ELSE 999 END",
"ELSE 999 END, m.protocol ASC",
"GROUP_CONCAT(CAST(COALESCE(m.daily_mileage_km, 0) AS CHAR) ORDER BY",
@@ -526,6 +527,7 @@ func TestMileageQueriesRespectEnabledProtocolPriority(t *testing.T) {
statistics := buildMileageStatisticsSummarySQL(query)
for _, want := range []string{
"m.protocol IN (?,?)",
"CASE WHEN COALESCE(m.daily_mileage_km, 0) > 0 THEN 0 ELSE 1 END",
"CASE m.protocol WHEN 'JT808' THEN 1 WHEN 'GB32960' THEN 2 ELSE 999 END",
"SUBSTRING_INDEX(GROUP_CONCAT",
} {
@@ -534,7 +536,7 @@ func TestMileageQueriesRespectEnabledProtocolPriority(t *testing.T) {
}
}
if strings.Contains(statistics.Text, "MAX(COALESCE(m.daily_mileage_km, 0))") {
t.Fatalf("configured priority must select the preferred source instead of the maximum mileage: %s", statistics.Text)
t.Fatalf("configured priority must select the preferred usable source instead of the maximum mileage: %s", statistics.Text)
}
fleet := buildFleetLatestMileageSQL(query)
@@ -543,6 +545,24 @@ func TestMileageQueriesRespectEnabledProtocolPriority(t *testing.T) {
}
}
func TestMileagePriorityFallsThroughZeroOnlyWhenAnotherSourceMoved(t *testing.T) {
query := url.Values{
"protocols": {"YUTONG_MQTT,JT808"},
"dateFrom": {"2026-07-19"},
"dateTo": {"2026-07-19"},
"deduplicate": {"1"},
}
daily := buildDailyMileageSQL(query)
effective := strings.Index(daily.Text, "CASE WHEN COALESCE(m.daily_mileage_km, 0) > 0 THEN 0 ELSE 1 END")
priority := strings.Index(daily.Text, "CASE m.protocol WHEN 'YUTONG_MQTT' THEN 1 WHEN 'JT808' THEN 2 ELSE 999 END")
if effective < 0 || priority < 0 || effective > priority {
t.Fatalf("effective mileage must be evaluated before configured protocol priority: %s", daily.Text)
}
if !strings.Contains(daily.Text, "GROUP BY m.vin, m.stat_date") {
t.Fatalf("zero fallback must remain scoped to one vehicle day: %s", daily.Text)
}
}
func TestMileageProtocolsRejectUnknownValues(t *testing.T) {
protocols := parseMileageProtocols("JT808,invalid');DROP TABLE vehicle_daily_mileage;--,JT808,YUTONG_MQTT")
if fmt.Sprint(protocols) != "[JT808 YUTONG_MQTT]" {

4
vehicle-data-platform/apps/web/src/v2/pages/StatisticsPage.tsx
View File

@@ -210,7 +210,7 @@ function SourceStrategy({ value, onChange }: { value: MileageSourceOption[]; onC
width={mobileLayout ? undefined : 430}
height={mobileLayout ? 'min(58dvh, 500px)' : undefined}
title="数据源策略"
description="同车同日按优先级选择一个里程来源"
description="同车同日优先采用靠前且有有效增量的里程来源"
badge={`${enabled.length}/3 已启用`}
badgeColor={enabled.length === 3 ? 'green' : 'blue'}
summaryItems={[
@@ -219,7 +219,7 @@ function SourceStrategy({ value, onChange }: { value: MileageSourceOption[]; onC
{ label: '统计口径', value: '单车单日', detail: '仅采用一个优先来源' }
]}
onCancel={() => setOpen(false)}
footerNote="避免多协议重复累计；至少保留一个来源。"
footerNote="首选来源为 0、其他来源有有效增量时会自动降级；全部为 0 时仍按优先级选择。"
primaryAction={{ label: '完成', onClick: () => setOpen(false) }}
>
<div className="v2-mileage-source-list">{value.map((source, index) => <Card key={source.protocol} className={`v2-mileage-source-card${source.enabled ? '' : ' is-disabled'}`} bodyStyle={{ padding: 0 }}>