feat(mileage): 接入区间汇总并复用查询结果
ci/woodpecker/push/woodpecker Pipeline was successful

This commit is contained in:
shishengliang
2026-10-29 09:11:21 +08:00
parent 8afe13c9a3
commit e2f932232f
33 changed files with 1056 additions and 172 deletions
+17 -12
View File
@@ -3,16 +3,16 @@ import { fileURLToPath } from 'node:url';
import { dirname, join } from 'node:path';
import pool from '../../db.js';
import { fetchVehicleInfoMap } from './vehicle-info.js';
import { fetchOneOsDailyMileage, fetchOneOsMileageDates } from './oneos-api.js';
import type { OneOsProtocol } from './source-policy.js';
import { buildRangeMileage, type RangeMileageResult } from './range-model.js';
import { fetchOneOsDailyMileage } from './oneos-api.js';
import { queryMonitoringRange } from './query-cache.js';
import { DEFAULT_ONEOS_PROTOCOL_PRIORITY, type OneOsProtocol } from './source-policy.js';
import { buildSummaryMileage } from './summary-model.js';
import {
buildMonitoringFilters,
buildPlateTargetNamesMap,
buildTargetPlatesMap,
dailyMileageMap,
mergeMonitoringVehicles,
mileageDatesBetween,
previousMileageDate,
toMileageRows,
type TargetRow,
@@ -117,17 +117,22 @@ export async function queryDateMileage(
export async function queryRangeMileage(
startDate: string,
endDate: string,
protocolPriority?: OneOsProtocol[],
): Promise<RangeMileageResult> {
const days = mileageDatesBetween(startDate, endDate);
const [apiRowsByDate, endDateRows, yesterdayRows, infoMap, targetRows] = await Promise.all([
fetchOneOsMileageDates(days, undefined, protocolPriority),
fetchOneOsDailyMileage(endDate, undefined, protocolPriority),
fetchOneOsDailyMileage(previousMileageDate(startDate), undefined, protocolPriority),
protocolPriority: OneOsProtocol[] = DEFAULT_ONEOS_PROTOCOL_PRIORITY,
refresh = false,
) {
const [cached, infoMap, targetRows] = await Promise.all([
queryMonitoringRange(startDate, endDate, protocolPriority, refresh),
fetchVehicleInfoMap(),
fetchTargetRows(),
]);
return buildRangeMileage(startDate, endDate, apiRowsByDate, endDateRows, yesterdayRows, infoMap, targetRows, regionMap);
const { summary, yesterdayRows } = cached.result;
return {
vehicles: buildSummaryMileage(summary, yesterdayRows, infoMap, targetRows, regionMap),
summary,
start: startDate, end: endDate,
queriedAt: cached.queriedAt,
cacheStatus: cached.cacheStatus,
};
}
export function buildDateFilters(vehicles: CachedVehicle[]): MonitoringFilters {
@@ -0,0 +1,82 @@
import assert from 'node:assert/strict';
import test from 'node:test';
import { Hono } from 'hono';
import pool from '../../db.js';
import type { AuthUser } from '../../auth/types.js';
import type { MonitoringResponse } from './types.js';
import monitoring from './monitoring.js';
import recent from './vehicle-recent.js';
const originalConfig = { url: process.env.ONEOS_MILEAGE_API_BASE_URL, key: process.env.ONEOS_MILEAGE_API_KEY };
test.before(() => { process.env.ONEOS_MILEAGE_API_BASE_URL = 'https://oneos.test'; process.env.ONEOS_MILEAGE_API_KEY = 'test-only'; });
test.after(() => {
if (originalConfig.url === undefined) delete process.env.ONEOS_MILEAGE_API_BASE_URL; else process.env.ONEOS_MILEAGE_API_BASE_URL = originalConfig.url;
if (originalConfig.key === undefined) delete process.env.ONEOS_MILEAGE_API_KEY; else process.env.ONEOS_MILEAGE_API_KEY = originalConfig.key;
});
test('summary integration preserves scope, valid partial mileage, trend nulls, reuse, details and refresh', async t => {
const requests: {url:string;body:any}[]=[];
let reassigned=false;
let failure=0;
t.mock.method(globalThis,'fetch',async(url:string,options:RequestInit)=>{
const body=JSON.parse(options.body as string); requests.push({url,body});
if(url.endsWith('/summary/query')){
if(failure)return Response.json({code:'UPSTREAM_ERROR',traceId:'trace'}, {status:failure,headers:{'Retry-After':'12'}});
const selected=body.plateNumbers || ['TEST0','TEST1','TEST2'];
const values=[0,30,20];
return Response.json({code:'SUCCESS',startDate:body.startDate,endDate:body.endDate,timezone:'Asia/Shanghai',vehicleCount:selected.length,
data:{vehicles:selected.map((plate:string)=>({plateNumber:plate,vin:'VIN',periodMileageKm:values[Number(plate.slice(-1))],
endDateTotalMileageKm:plate==='TEST0'?null:100,endDateDataTime:null,endDateSourceProtocol:null,
validDays:1,missingDays:plate==='TEST2'?0:1,anomalyDays:plate==='TEST2'?1:0,
status:plate==='TEST2'?'DATA_ANOMALY':'PARTIAL_DATA',qualityReasons:[plate==='TEST2'?'BASELINE_MISSING':'CARRIED_FORWARD'],sourceProtocols:['MQTT'],updatedAt:null})),
dailyTrend:[{date:body.startDate,mileageKm:selected.reduce((sum:number,p:string)=>sum+values[Number(p.slice(-1))],0),validVehicleCount:selected.length,missingVehicleCount:0,anomalyVehicleCount:0},
{date:body.endDate,mileageKm:null,validVehicleCount:0,missingVehicleCount:selected.filter((p:string)=>p!=='TEST2').length,anomalyVehicleCount:selected.includes('TEST2')?1:0}]},
snapshotId:'snapshot',generatedAt:'2026-10-27T00:00:00Z',expiresAt:'2099-01-01T00:00:00Z',nextCursor:null});
}
if(url.endsWith('/range/query'))return Response.json({code:'SUCCESS',data:(body.plateNumbers||[]).flatMap((plate:string)=>[
{plateNumber:plate,date:body.startDate,dailyMileageKm:20,totalMileageKm:100,status:'NORMAL',sourceProtocol:'MQTT'},
{plateNumber:plate,date:body.endDate,dailyMileageKm:0,totalMileageKm:100,status:'NORMAL',dataQuality:'CARRIED_FORWARD',sourceProtocol:'MQTT'},
]),snapshotId:'detail',nextCursor:null});
return Response.json({code:'SUCCESS',data:[]});
});
t.mock.method(pool,'execute',async(sql:string)=>{
if(!sql.includes('FROM vehicle_info'))return [[],[]];
return [[0,1,2,3].map(i=>({plate:`TEST${i}`,vin:'VIN',customer:'客户',department:i===1?'B':'A',manager:null,
manager_id:i===0&&reassigned?'new-owner':`${i}`,rent_status:'租赁',entity:null,project:null,inventory_location:null,brand_label:null})),[]];
});
let user:AuthUser={userId:'0',userName:'测试',loginName:'test',depCode:'',depName:'A',roles:[],permissionLevel:'full'};
const app=new Hono<{Variables:{user:AuthUser}}>();
app.use('*',async(c,next)=>{c.set('user',user);await next();});
app.route('/monitoring',monitoring);app.route('/vehicle',recent);
const base='/monitoring?startDate=2026-10-10&endDate=2026-10-11&sourcePriority=instrument';
const read=async(extra='')=>{const response=await app.request(base+extra);assert.equal(response.status,200);return await response.json() as MonitoringResponse;};
const first=await read('&limit=1');
assert.equal(first.vehicles[0].plate,'TEST1');assert.equal(first.stats.totalToday,50);
assert.equal(first.rangeDailyTotals?.[1].totalKm,null);
assert.equal(requests.filter(r=>r.url.endsWith('/summary/query')).length,1);
assert.equal(requests.filter(r=>r.url.endsWith('/range/query')).length,0);
const second=await read('&page=2&limit=1&includeDailyTrend=0');assert.equal(second.vehicles[0].periodMileageKm,20);
const sorted=await read('&sortOrder=asc&includeDailyTrend=0');assert.equal(sorted.vehicles[0].periodMileageKm,0);
const exported=await read('&limit=9999&includeDailyTrend=0');
assert.equal(exported.vehicles.length,4);assert.equal(exported.vehicles[3].periodMileageKm,null);
assert.equal(exported.queriedAt,first.queriedAt);assert.ok(!exported.vehicles[0].dailyMileage);
assert.equal(requests.filter(r=>r.url.endsWith('/summary/query')).length,1);
user={...user,permissionLevel:'department'};
const department=await read();assert.equal(department.stats.totalToday,20);
assert.equal(department.rangeDailyTotals?.[0].totalKm,20);
assert.deepEqual(requests.filter(r=>r.url.endsWith('/summary/query')).at(-1)!.body.plateNumbers,['TEST0','TEST2']);
user={...user,permissionLevel:'personal'};
const personal=await read();assert.equal(personal.rangeDailyTotals?.[0].totalKm,0);
assert.equal(personal.rangeDailyTotals?.[1].totalKm,null);
const count=requests.length;reassigned=true;
const empty=await read();assert.equal(empty.total,0);assert.ok(empty.rangeDailyTotals?.every(d=>d.totalKm===null));assert.equal(requests.length,count);
const denied=await app.request('/vehicle/TEST0/recent?start=2026-10-10&end=2026-10-11');assert.equal(denied.status,403);
reassigned=false;user={...user,permissionLevel:'full'};
const detail=await read('&plate=TEST2&dailyDetails=1&includeDailyTrend=0');
assert.deepEqual(detail.vehicles[0].dailyMileage,{'2026-10-10':20,'2026-10-11':null});
assert.deepEqual(requests.at(-1)!.body.plateNumbers,['TEST2']);
const before=requests.filter(r=>r.url.endsWith('/summary/query')).length;
await read('&refresh=1');assert.equal(requests.filter(r=>r.url.endsWith('/summary/query')).length,before+1);
failure=403;const fail=await app.request(base+'&refresh=1');assert.equal(fail.status,403);assert.equal((await fail.json()).code,'UPSTREAM_ERROR');
failure=429;const limited=await app.request(base);assert.equal(limited.status,429);assert.equal(limited.headers.get('Retry-After'),'12');
});
+54 -10
View File
@@ -1,4 +1,9 @@
import { Hono } from 'hono';
import { queryMonitoringRange } from './query-cache.js';
import { MileageSummaryError, type MileageSummary } from './summary-api.js';
import { mileageDatesBetween } from './cache-model.js';
import { fetchOneOsMileageDates } from './oneos-api.js';
import { validDailyMileage, type OneOsDailyMileage } from './oneos-model.js';
import { getCache, queryDateMileage, queryRangeMileage, buildDateFilters } from './cache.js';
import { filterByPermission, maskCustomerNames } from '../../auth/permissions.js';
import type { AuthUser } from '../../auth/types.js';
@@ -132,21 +137,29 @@ app.get('/', async (c) => {
brands: c.req.queries('brand') || [],
};
let rangeSummary: MileageSummary | undefined;
let queriedAt: string | undefined;
let allVehicles: CachedVehicle[];
let filters: MonitoringFilters;
let rangeDailyTotals: { date: string; totalKm: number }[] | undefined;
let rangeDailyTotals: { date: string; totalKm: number | null }[] | undefined;
let dateRange: { start: string; end: string } | undefined;
if (range) {
try {
const result = await queryRangeMileage(range.start, range.end, protocolPriority);
const result = await queryRangeMileage(range.start, range.end, protocolPriority, c.req.query('refresh') === '1');
queriedAt = result.queriedAt;
c.header('X-Mileage-Cache', result.cacheStatus);
allVehicles = result.vehicles;
rangeDailyTotals = result.dailyTotals;
rangeSummary = result.summary;
dateRange = { start: result.start, end: result.end };
filters = buildDateFilters(allVehicles);
} catch (e: unknown) {
if (e instanceof MileageSummaryError) {
if (e.retryAfter) c.header('Retry-After', e.retryAfter);
return c.json({ error: e.message, code: e.code, traceId: e.traceId }, e.status === 401 ? 502 : e.status as 400 | 403 | 429 | 500 | 502 | 504);
}
console.error('monitoring range query error:', e);
return c.json(EMPTY_RESPONSE, 500);
return c.json({ error: e instanceof Error ? e.message : '里程汇总查询失败' }, 502);
}
} else if (date) {
try {
@@ -177,17 +190,35 @@ app.get('/', async (c) => {
}
const filtered = applyFilters(allVehicles, filterParams);
if (rangeDailyTotals && filtered.length !== allVehicles.length) {
rangeDailyTotals = rangeDailyTotals.map(item => ({
...item,
totalKm: Math.round(filtered.reduce((sum, vehicle) => sum + (vehicle.dailyMileage?.[item.date] || 0), 0)),
}));
// The provider trend covers its entire selected fleet, not the BI user's scope.
if (range && rangeSummary && c.req.query('includeDailyTrend') !== '0') {
const authorized = new Set(rangeSummary.vehicles.map(v => v.plateNumber));
const plates = [...new Set(filtered.map(v => v.plate.trim().toUpperCase()).filter(p => authorized.has(p)))].sort();
if (plates.length === 0) {
rangeDailyTotals = mileageDatesBetween(range.start, range.end).map(date => ({ date, totalKm: null }));
} else if (range.start === range.end) {
const valid = filtered.filter(v => v.periodMileageKm != null);
rangeDailyTotals = [{ date: range.start, totalKm: valid.length ? valid.reduce((sum,v) => sum + v.periodMileageKm!, 0) : null }];
} else {
try {
const summary = plates.length === authorized.size ? rangeSummary
: (await queryMonitoringRange(range.start, range.end, protocolPriority, c.req.query('refresh') === '1', plates)).result.summary;
rangeDailyTotals = summary.dailyTrend.map(d => ({ date: d.date, totalKm: d.mileageKm }));
} catch (e) {
if (e instanceof MileageSummaryError) {
if (e.retryAfter) c.header('Retry-After', e.retryAfter);
return c.json({ error: e.message, code: e.code, traceId: e.traceId }, e.status === 401 ? 502 : e.status as 400 | 403 | 429 | 500 | 502 | 504);
}
return c.json({ error: '所选车辆趋势查询失败,请重试' }, 502);
}
}
}
const stats = {
totalToday: Math.round(filtered.reduce((sum, v) => sum + v.dailyKm, 0)),
totalAll: filtered.reduce((sum, v) => sum + (v.totalKm || 0), 0),
vehicleCount: filtered.length,
...(rangeSummary ? { validVehicleCount: filtered.filter(v => v.periodMileageKm != null).length, totalMileageVehicleCount: filtered.filter(v => v.totalKm != null).length } : {}),
yesterdayTotal: filtered.reduce((sum, v) => sum + v.yesterdayKm, 0),
};
@@ -215,7 +246,19 @@ app.get('/', async (c) => {
});
const offset = (page - 1) * limit;
const paged = sorted.slice(offset, offset + limit);
let paged = sorted.slice(offset, offset + limit);
if (range && rangeSummary && c.req.query('dailyDetails') === '1') {
const authorized = new Set(rangeSummary.vehicles.map(v => v.plateNumber));
const plates = paged.map(v => v.plate.trim().toUpperCase()).filter(p => authorized.has(p));
try {
const dates = mileageDatesBetween(range.start, range.end);
const rows = plates.length ? await fetchOneOsMileageDates(dates, plates, protocolPriority) : new Map<string, OneOsDailyMileage[]>();
const byDate = new Map(dates.map(date => [date, new Map<string, number | null>((rows.get(date) || []).map(row => [row.plateNumber, validDailyMileage(row)]))]));
paged = paged.map(v => ({ ...v, dailyMileage: Object.fromEntries(dates.map(date => [date, byDate.get(date)?.get(v.plate.trim().toUpperCase()) ?? null])) }));
} catch {
return c.json({ error: '逐日明细导出失败,请缩小范围或重试' }, 502);
}
}
const total = filtered.length;
return c.json({
@@ -229,6 +272,7 @@ app.get('/', async (c) => {
totalPages: Math.ceil(total / limit),
updatedAt: dateRange?.end || date || getCache()?.updatedAt || new Date().toISOString(),
sourcePriority,
queriedAt,
});
});
+2 -1
View File
@@ -224,6 +224,7 @@ export async function fetchOneOsDailyMileage(
date: string,
plateNumbers?: string[],
protocolPriority: OneOsProtocol[] = DEFAULT_ONEOS_PROTOCOL_PRIORITY,
refresh = false,
): Promise<OneOsDailyMileage[]> {
assertMileageDate(date);
@@ -231,7 +232,7 @@ export async function fetchOneOsDailyMileage(
const cacheKey = `${date}:${policyKey}`;
const hit = cache.get(cacheKey);
let rows: OneOsDailyMileage[];
if (hit && hit.expiresAt > Date.now()) {
if (!refresh && hit && hit.expiresAt > Date.now()) {
rows = hit.rows;
} else {
let pending = inflight.get(cacheKey);
@@ -98,3 +98,13 @@ test('异常日里程不伪造为有效值,保留仪表累计和异常原因',
assert.equal(row.totalMileageKm, 66852.4);
assert.equal(row.dataQuality, 'ODOMETER_SOURCE_CHANGED');
});
test('NORMAL carried-forward detail is missing while actual zero remains valid', async () => {
const { validDailyMileage } = await import('./oneos-model.js');
const rows=normalizeOneOsRows([
{plateNumber:'MISSING',status:'NORMAL',dailyMileageKm:0,dataQuality:'CARRIED_FORWARD'},
{plateNumber:'ZERO',status:'NORMAL',dailyMileageKm:0},
],'2026-10-01');
assert.equal(validDailyMileage(rows[0]),null);
assert.equal(validDailyMileage(rows[1]),0);
});
+2 -2
View File
@@ -17,7 +17,7 @@ export interface OneOsDailyMileage {
/** Missing or unusable daily mileage must not contribute to interval statistics. */
export function validDailyMileage(row: OneOsDailyMileage | undefined): number | null {
const km = row?.dailyMileageKm;
return row?.status === 'NORMAL' && km != null && Number.isFinite(km) && km >= 0 ? km : null;
return row?.status === 'NORMAL' && row.dataQuality !== 'CARRIED_FORWARD' && km != null && Number.isFinite(km) && km >= 0 ? km : null;
}
const MILEAGE_DATE_PATTERN = /^\d{4}-\d{2}-\d{2}$/;
@@ -69,7 +69,7 @@ export function normalizeOneOsRows(value: unknown, requestedDate: string): OneOs
dailyMileageKm: nonNegativeMileage(row.dailyMileageKm, status),
totalMileageKm: nonNegativeMileage(row.totalMileageKm, status === 'DATA_ANOMALY' ? 'NORMAL' : status),
status,
...(status === 'DATA_ANOMALY' ? { dataQuality: stringField(row, 'dataQuality') || 'DATA_ANOMALY' } : {}),
...((row.dataQuality || status === 'DATA_ANOMALY') ? { dataQuality: stringField(row, 'dataQuality') || 'DATA_ANOMALY' } : {}),
dataTime,
calculatedAt,
updatedAt,
@@ -0,0 +1,89 @@
import assert from 'node:assert/strict';
import test from 'node:test';
import { createMonitoringQueryCache, type RangeMileageSource } from './query-cache.js';
import type { OneOsProtocol } from './source-policy.js';
const protocols: OneOsProtocol[] = ['GB32960', 'MQTT'];
const result = (): RangeMileageSource => ({ summary: { vehicles: [], dailyTrend: [], snapshotId: 'test', generatedAt: '2026-01-01T00:00:00Z', expiresAt: '2099-01-01T00:00:00Z' }, yesterdayRows: [] });
test('completed range results are reused for five minutes; refresh bypasses daily caches too', async () => {
let time = 1_000;
const calls: boolean[] = [];
const query = createMonitoringQueryCache(async (start, end, _protocols, refresh) => {
calls.push(!!refresh);
return result();
}, () => time);
const first = await query('2026-10-01', '2026-10-26', protocols);
assert.equal(first.cacheStatus, 'miss');
time += 299_999;
const hit = await query('2026-10-01', '2026-10-26', protocols);
assert.equal(hit.cacheStatus, 'hit');
assert.equal(hit.result, first.result);
assert.equal(hit.queriedAt, first.queriedAt);
time += 1;
assert.equal((await query('2026-10-01', '2026-10-26', protocols)).cacheStatus, 'miss');
assert.equal((await query('2026-10-01', '2026-10-26', protocols, true)).cacheStatus, 'refresh');
assert.deepEqual(calls, [false, false, true]);
});
test('date range and ordered protocol priority isolate cached results; entries are bounded', async () => {
let calls = 0;
const query = createMonitoringQueryCache(async (start, end) => { calls++; return result(); });
await query('2026-10-01', '2026-10-26', protocols);
await query('2026-10-01', '2026-10-26', ['MQTT', 'GB32960']);
await query('2026-10-01', '2026-10-26', ['JT808']);
await query('2026-10-02', '2026-10-26', protocols);
assert.equal(calls, 4);
assert.equal((await query('2026-10-01', '2026-10-26', protocols)).cacheStatus, 'hit');
await query('2026-10-03', '2026-10-26', protocols);
assert.equal((await query('2026-10-01', '2026-10-26', protocols)).cacheStatus, 'miss');
assert.equal(calls, 6);
});
test('simultaneous requests share upstream mileage retrieval, including explicit refresh', async () => {
let calls = 0;
let release!: (value: RangeMileageSource) => void;
const query = createMonitoringQueryCache(async () => {
calls++;
return new Promise<RangeMileageSource>(resolve => { release = resolve; });
});
const first = query('2026-10-01', '2026-10-26', protocols);
const second = query('2026-10-01', '2026-10-26', protocols);
const refresh = query('2026-10-01', '2026-10-26', protocols, true);
assert.equal(calls, 1);
release(result());
const entries = await Promise.all([first, second, refresh]);
assert.deepEqual(entries.map(entry => entry.cacheStatus), ['miss', 'shared', 'shared']);
assert.ok(entries.every(entry => entry.result === entries[0].result));
});
test('failed refresh removes stale data and failed requests can be retried', async () => {
let fail = false;
let calls = 0;
const query = createMonitoringQueryCache(async (start, end) => {
calls++;
if (fail) throw new Error('upstream unavailable');
return result();
});
await query('2026-10-01', '2026-10-26', protocols);
fail = true;
await assert.rejects(query('2026-10-01', '2026-10-26', protocols, true), /upstream unavailable/);
await assert.rejects(query('2026-10-01', '2026-10-26', protocols), /upstream unavailable/);
fail = false;
assert.equal((await query('2026-10-01', '2026-10-26', protocols)).cacheStatus, 'miss');
assert.equal(calls, 4);
});
test('summary cache isolates normalized vehicle scopes and respects provider snapshot expiry', async () => {
let time=1_000,calls=0;
const query=createMonitoringQueryCache(async()=>{
calls++;
const value=result();value.summary.expiresAt=new Date(2_000).toISOString();return value;
},()=>time);
await query('2026-10-01','2026-10-26',protocols,false,[' b ','A']);
assert.equal((await query('2026-10-01','2026-10-26',protocols,false,['A','B'])).cacheStatus,'hit');
await query('2026-10-01','2026-10-26',protocols,false,['A']);
assert.equal(calls,2);
time=2_000;
assert.equal((await query('2026-10-01','2026-10-26',protocols,false,['A','B'])).cacheStatus,'miss');
});
+58
View File
@@ -0,0 +1,58 @@
import { fetchOneOsDailyMileage, type OneOsDailyMileage } from './oneos-api.js';
import { previousMileageDate } from './cache-model.js';
import { fetchMileageSummary, type MileageSummary } from './summary-api.js';
import type { OneOsProtocol } from './source-policy.js';
// Cache upstream mileage only; ownership, permissions and business filters stay live.
const TTL_MS = 5 * 60 * 1000;
const MAX_ENTRIES = 4;
export interface RangeMileageSource {
summary: MileageSummary;
yesterdayRows: OneOsDailyMileage[];
}
type Entry = { result: RangeMileageSource; queriedAt: string; expiresAt: number };
async function loadSource(start: string, end: string, protocols: OneOsProtocol[], refresh = false, plates?: string[]): Promise<RangeMileageSource> {
const [summary, yesterdayRows] = await Promise.all([
fetchMileageSummary(start, end, protocols, plates),
plates ? Promise.resolve([]) : fetchOneOsDailyMileage(previousMileageDate(start), undefined, protocols, refresh),
]);
return { summary, yesterdayRows };
}
export function createMonitoringQueryCache(
query = loadSource,
now: () => number = Date.now,
) {
const entries = new Map<string, Entry>();
const inflight = new Map<string, Promise<Entry>>();
return async (start: string, end: string, protocols: OneOsProtocol[], refresh = false, plates?: string[]) => {
const selected = plates ? [...new Set(plates.map(p => p.trim().toUpperCase()))].sort() : undefined;
const key = JSON.stringify([start, end, protocols, selected]);
const timestamp = now();
for (const [entryKey, entry] of entries) {
if (entry.expiresAt <= timestamp) entries.delete(entryKey);
}
const pending = inflight.get(key);
if (pending) return { ...await pending, cacheStatus: 'shared' as const };
const hit = entries.get(key);
if (hit && !refresh) return { ...hit, cacheStatus: 'hit' as const };
// A failed explicit refresh must not silently fall back to the old result.
entries.delete(key);
const request = query(start, end, protocols, refresh, selected).then(result => {
const completedAt = now();
const entry = { result, queriedAt: new Date(completedAt).toISOString(), expiresAt: Math.min(completedAt + TTL_MS, Date.parse(result.summary.expiresAt)) };
entries.set(key, entry);
if (entries.size > MAX_ENTRIES) entries.delete(entries.keys().next().value!);
return entry;
}).finally(() => inflight.delete(key));
inflight.set(key, request);
return { ...await request, cacheStatus: refresh ? 'refresh' as const : 'miss' as const };
};
}
export const queryMonitoringRange = createMonitoringQueryCache();
@@ -0,0 +1,74 @@
import assert from 'node:assert/strict';
import test from 'node:test';
import { fetchMileageSummary, parseMileageSummary, MileageSummaryError } from './summary-api.js';
const originalConfig = { url: process.env.ONEOS_MILEAGE_API_BASE_URL, key: process.env.ONEOS_MILEAGE_API_KEY };
test.before(() => { process.env.ONEOS_MILEAGE_API_BASE_URL = 'https://oneos.test'; process.env.ONEOS_MILEAGE_API_KEY = 'test-only'; });
test.after(() => {
if (originalConfig.url === undefined) delete process.env.ONEOS_MILEAGE_API_BASE_URL; else process.env.ONEOS_MILEAGE_API_BASE_URL = originalConfig.url;
if (originalConfig.key === undefined) delete process.env.ONEOS_MILEAGE_API_KEY; else process.env.ONEOS_MILEAGE_API_KEY = originalConfig.key;
});
const start = '2026-10-01', end = '2026-10-02';
const vehicle = (plate: string) => ({ plateNumber: plate, vin: 'VIN', periodMileageKm: 0, endDateTotalMileageKm: null,
endDateDataTime: null, endDateSourceProtocol: null, validDays: 1, missingDays: 1, anomalyDays: 0,
status: 'PARTIAL_DATA', qualityReasons: ['CARRIED_FORWARD'], sourceProtocols: ['MQTT'], updatedAt: null });
const payload = () => ({ code: 'SUCCESS', startDate: start, endDate: end, timezone: 'Asia/Shanghai', vehicleCount: 1,
data: { vehicles: [vehicle('TEST0')], dailyTrend: [
{ date: start, mileageKm: 0, validVehicleCount: 1, missingVehicleCount: 0, anomalyVehicleCount: 0 },
{ date: end, mileageKm: null, validVehicleCount: 0, missingVehicleCount: 1, anomalyVehicleCount: 0 },
] }, snapshotId: 's1', generatedAt: '2026-10-27T00:00:00Z', expiresAt: '2099-01-01T00:00:00Z', nextCursor: null as string | null });
test('summary boundary preserves real zero and missing trend; rejects malformed counts/status/values', () => {
const p = payload();
assert.equal(parseMileageSummary(p,start,end).vehicles[0].periodMileageKm,0);
assert.equal(parseMileageSummary(p,start,end).dailyTrend[1].mileageKm,null);
for (const mutate of [
(p: ReturnType<typeof payload>) => { p.data.vehicles[0].validDays = 2; },
(p: ReturnType<typeof payload>) => { p.data.vehicles[0].status = 'NORMAL'; },
(p: ReturnType<typeof payload>) => { p.data.vehicles[0].periodMileageKm = NaN; },
(p: ReturnType<typeof payload>) => { p.data.dailyTrend[1].mileageKm = 0; },
]) { const bad = payload(); mutate(bad); assert.throws(() => parseMileageSummary(bad,start,end)); }
});
test('summary pagination keeps the snapshot and reuses one fleet trend; empty scope never means all', async t => {
t.mock.method(globalThis, 'fetch', async (_url: string, options: RequestInit) => {
const body = JSON.parse(options.body as string);
assert.deepEqual(body.plateNumbers, ['TEST0','TEST1']);
assert.equal(body.includeDailyTrend,true);
assert.equal(body.pageSize,5000);
const p = payload(); p.vehicleCount = 2;
p.data.vehicles = [vehicle(body.cursor ? 'TEST1':'TEST0')];
p.data.dailyTrend[0].validVehicleCount = 2;
p.data.dailyTrend[1].missingVehicleCount = 2;
p.nextCursor = body.cursor ? null:'next';
return Response.json(p);
});
const result = await fetchMileageSummary(start,end,['MQTT'],['TEST0','TEST1']);
assert.equal(result.vehicles.length,2);
assert.equal(result.dailyTrend.length,2);
assert.equal(result.dailyTrend[1].mileageKm,null);
await assert.rejects(fetchMileageSummary(start,end,['MQTT'],[]), /空车牌/);
});
test('summary authorization, throttling and timeout errors are explicit and never return empty success', async t => {
for (const status of [401,403,429,504]) {
const mock = t.mock.method(globalThis, 'fetch', async () => Response.json({code:'TEST_ERROR',traceId:'test-trace'}, {status,headers:{'Retry-After':'30'}}));
await assert.rejects(fetchMileageSummary(start,end,['MQTT']), e => e instanceof MileageSummaryError && e.status === status && e.traceId === 'test-trace' && e.retryAfter === '30');
mock.mock.restore();
}
});
test('snapshot changes and duplicate vehicles reject partial pagination', async t => {
for (const changedSnapshot of [true,false]) {
const mock=t.mock.method(globalThis,'fetch',async (_url:string,options:RequestInit)=>{
const body=JSON.parse(options.body as string), p=payload(); p.vehicleCount=2;
p.data.dailyTrend[0].validVehicleCount=2; p.data.dailyTrend[1].missingVehicleCount=2;
p.nextCursor=body.cursor?null:'next';
if(body.cursor && changedSnapshot)p.snapshotId='s2';
return Response.json(p);
});
await assert.rejects(fetchMileageSummary(start,end,['MQTT']),changedSnapshot?/快照/:/重复车辆/);
mock.mock.restore();
}
});
+116
View File
@@ -0,0 +1,116 @@
import dotenv from 'dotenv';
import { mileageDatesBetween } from './cache-model.js';
import type { OneOsProtocol } from './source-policy.js';
dotenv.config();
export interface MileageSummaryVehicle {
vin: string;
plateNumber: string;
periodMileageKm: number | null;
endDateTotalMileageKm: number | null;
endDateDataTime: string | null;
endDateSourceProtocol: OneOsProtocol | null;
validDays: number;
missingDays: number;
anomalyDays: number;
status: 'NORMAL' | 'PARTIAL_DATA' | 'NO_DATA' | 'DATA_ANOMALY';
qualityReasons: string[];
sourceProtocols: OneOsProtocol[];
updatedAt: string | null;
}
export interface MileageSummaryTrend {
date: string;
mileageKm: number | null;
validVehicleCount: number;
missingVehicleCount: number;
anomalyVehicleCount: number;
}
export interface MileageSummary {
vehicles: MileageSummaryVehicle[];
dailyTrend: MileageSummaryTrend[];
snapshotId: string;
generatedAt: string;
expiresAt: string;
}
export class MileageSummaryError extends Error {
constructor(readonly status: number, readonly code: string, readonly traceId?: string, readonly retryAfter?: string) {
const message = status === 401 ? '里程服务凭据无效或未覆盖所选日期'
: status === 403 ? '里程服务未授权所选车辆或完整日期区间'
: status === 429 ? `里程查询频率受限,请${retryAfter ? ` ${retryAfter} 秒` : '稍'}后重试`
: status === 504 ? '里程查询超时,请缩小车辆或日期范围后重试'
: status === 400 ? '里程查询参数或快照无效,请重新查询'
: '里程服务暂不可用,请重试';
super(message);
}
}
// Validate once at the provider boundary; invalid responses must not become zero mileage.
export function parseMileageSummary(value: unknown, start: string, end: string): MileageSummary & { nextCursor: string | null; vehicleCount: number } {
const p = value as any;
const dates = mileageDatesBetween(start, end);
const protocols = new Set(['GB32960', 'MQTT', 'JT808']);
const count = (n: unknown) => Number.isInteger(n) && Number(n) >= 0;
const km = (n: unknown) => n === null || typeof n === 'number' && Number.isFinite(n) && n >= 0;
const time = (s: unknown) => s === null || typeof s === 'string' && Number.isFinite(Date.parse(s));
if (!p || p.code !== 'SUCCESS' || p.startDate !== start || p.endDate !== end || p.timezone !== 'Asia/Shanghai'
|| !count(p.vehicleCount) || p.vehicleCount > 5000 || !p.snapshotId || typeof p.snapshotId !== 'string'
|| typeof p.generatedAt !== 'string' || !time(p.generatedAt) || typeof p.expiresAt !== 'string' || !time(p.expiresAt)
|| !(p.nextCursor === null || typeof p.nextCursor === 'string' && p.nextCursor.length > 0)
|| !Array.isArray(p.data?.vehicles) || !Array.isArray(p.data?.dailyTrend)) throw new Error('里程汇总响应格式不符合契约');
for (const v of p.data.vehicles) {
if (!v || typeof v.plateNumber !== 'string' || !v.plateNumber.trim() || typeof v.vin !== 'string'
|| !km(v.periodMileageKm) || !km(v.endDateTotalMileageKm) || !time(v.endDateDataTime) || !time(v.updatedAt)
|| !(v.endDateSourceProtocol === null || protocols.has(v.endDateSourceProtocol))
|| ![v.validDays,v.missingDays,v.anomalyDays].every(count) || v.validDays + v.missingDays + v.anomalyDays !== dates.length
|| (v.validDays === 0) !== (v.periodMileageKm === null)
|| !Array.isArray(v.qualityReasons) || !v.qualityReasons.every((x: unknown) => typeof x === 'string')
|| !Array.isArray(v.sourceProtocols) || !v.sourceProtocols.every((x: string) => protocols.has(x))) throw new Error('里程汇总车辆字段不符合契约');
const status = v.anomalyDays > 0 ? 'DATA_ANOMALY' : v.validDays === 0 ? 'NO_DATA' : v.missingDays > 0 ? 'PARTIAL_DATA' : 'NORMAL';
if (v.status !== status) throw new Error('里程汇总质量状态不符合契约');
}
if (p.data.dailyTrend.length !== dates.length || p.data.dailyTrend.some((d: any, i: number) => !d || d.date !== dates[i]
|| !km(d.mileageKm) || ![d.validVehicleCount,d.missingVehicleCount,d.anomalyVehicleCount].every(count)
|| d.validVehicleCount + d.missingVehicleCount + d.anomalyVehicleCount !== p.vehicleCount
|| (d.validVehicleCount === 0) !== (d.mileageKm === null))) throw new Error('里程每日趋势不符合契约');
return { vehicles: p.data.vehicles, dailyTrend: p.data.dailyTrend, snapshotId: p.snapshotId,
generatedAt: p.generatedAt, expiresAt: p.expiresAt, nextCursor: p.nextCursor, vehicleCount: p.vehicleCount };
}
export async function fetchMileageSummary(start: string, end: string, protocols: OneOsProtocol[], plates?: string[]): Promise<MileageSummary> {
// An empty BI scope must never become the provider's "all authorized vehicles".
if (plates?.length === 0) throw new Error('不能使用空车牌集合查询里程汇总');
const base = (process.env.ONEOS_MILEAGE_API_BASE_URL || '').replace(/\/+$/, '');
const key = process.env.ONEOS_MILEAGE_API_KEY;
if (!base || !key) throw new Error('OneOS mileage API is not configured');
const vehicles: MileageSummaryVehicle[] = [];
const seen = new Set<string>();
let first: ReturnType<typeof parseMileageSummary> | undefined;
let cursor: string | null = null;
do {
const response = await fetch(`${base}/api/v1/vehicles/mileage/summary/query`, {
method: 'POST', headers: { Authorization: `Bearer ${key}`, 'Content-Type': 'application/json' },
body: JSON.stringify({ startDate: start, endDate: end, protocolPriority: protocols,
includeDailyTrend: true, pageSize: 5000, ...(plates ? { plateNumbers: plates } : {}), ...(cursor ? { cursor } : {}) }),
signal: AbortSignal.timeout(Math.max(15000, Number(process.env.ONEOS_MILEAGE_API_TIMEOUT_MS) || 20000)),
});
const payload = await response.json();
if (!response.ok || payload.code !== 'SUCCESS') throw new MileageSummaryError(response.status === 200 ? 502 : response.status,
payload.code || 'UNKNOWN', payload.traceId, response.headers.get('Retry-After') || undefined);
const page = parseMileageSummary(payload, start, end);
if (first && (page.snapshotId !== first.snapshotId || page.vehicleCount !== first.vehicleCount
|| page.generatedAt !== first.generatedAt || page.expiresAt !== first.expiresAt)) throw new Error('里程汇总分页快照发生变化,请重新查询');
first ??= page;
for (const v of page.vehicles) {
const plate = v.plateNumber.trim().toUpperCase();
if (seen.has(plate)) throw new Error('里程汇总分页出现重复车辆');
seen.add(plate);
vehicles.push({ ...v, plateNumber: plate });
}
if (vehicles.length > first.vehicleCount || (page.nextCursor && page.vehicles.length === 0)) throw new Error('里程汇总分页数量不一致');
cursor = page.nextCursor;
} while (cursor);
if (vehicles.length !== first!.vehicleCount) throw new Error('里程汇总未返回完整车辆集合');
return { ...first!, vehicles }; // Reuse one complete trend; never sum trends from pages.
}
@@ -0,0 +1,19 @@
import assert from 'node:assert/strict';
import test from 'node:test';
import { buildSummaryMileage } from './summary-model.js';
import type { MileageSummary } from './summary-api.js';
import type { VehicleInfoRow } from './types.js';
test('summary mapping preserves valid partial amounts, independent odometers and observation timestamps',()=>{
const info:VehicleInfoRow={plate:'TEST',vin:'VIN',customer:null,department:'A',manager:null,manager_id:'1',rent_status:null,entity:null,project:null,inventory_location:null,brand_label:null};
const summary:MileageSummary={snapshotId:'s',generatedAt:'2026-10-27T00:00:00Z',expiresAt:'2099-01-01T00:00:00Z',dailyTrend:[],vehicles:[{
plateNumber:'TEST',vin:'VIN',periodMileageKm:25,endDateTotalMileageKm:100,endDateDataTime:'2026-10-26T00:00:00Z',endDateSourceProtocol:'JT808',
validDays:1,missingDays:0,anomalyDays:1,status:'DATA_ANOMALY',qualityReasons:['BASELINE_MISSING'],sourceProtocols:['MQTT','JT808'],updatedAt:'2026-10-26T01:00:00Z',
}]};
const map=new Map([['TEST',info]]);
const row=buildSummaryMileage(summary,[],map,[],{})[0];
assert.equal(row.periodMileageKm,25);assert.equal(row.isDataSynced,true);assert.equal(row.sourceCategory,'MIXED');assert.equal(row.totalKm,100);
assert.equal(row.dataTime,summary.vehicles[0].endDateDataTime);assert.equal(row.updatedAt,summary.vehicles[0].updatedAt);assert.equal(row.calculatedAt,null,'query generation time is not a vehicle observation time');
summary.vehicles[0]={...summary.vehicles[0],periodMileageKm:null,validDays:0,anomalyDays:2};
const missing=buildSummaryMileage(summary,[],map,[],{})[0];assert.equal(missing.periodMileageKm,null);assert.equal(missing.totalKm,100);assert.equal(missing.isDataSynced,false);
});
@@ -0,0 +1,26 @@
import type { MileageSummary } from './summary-api.js';
import type { OneOsDailyMileage } from './oneos-model.js';
import { buildPlateTargetNamesMap, dailyMileageMap, mergeMonitoringVehicles, type TargetRow } from './cache-model.js';
import type { CachedVehicle, VehicleInfoRow } from './types.js';
import { sourceCategoryFromProtocol } from './source-policy.js';
export function buildSummaryMileage(summary: MileageSummary, yesterdayRows: OneOsDailyMileage[], infoMap: Map<string, VehicleInfoRow>, targets: TargetRow[], regions: Record<string, string>): CachedVehicle[] {
const byPlate = new Map(summary.vehicles.map(v => [v.plateNumber, v]));
return mergeMonitoringVehicles([], infoMap, dailyMileageMap(yesterdayRows), buildPlateTargetNamesMap(targets), regions).map(vehicle => {
const row = byPlate.get(vehicle.plate.trim().toUpperCase());
if (!row) return { ...vehicle, periodMileageKm: null };
const categories = new Set(row.sourceProtocols.map(sourceCategoryFromProtocol));
return { ...vehicle, vin: row.vin || vehicle.vin,
dailyKm: row.periodMileageKm ?? 0, periodMileageKm: row.periodMileageKm,
totalKm: row.endDateTotalMileageKm,
validDays: row.validDays, missingDays: row.missingDays, anomalyDays: row.anomalyDays,
qualityStatus: row.status, qualityReasons: row.qualityReasons, sourceProtocols: row.sourceProtocols,
mileageAnomaly: row.anomalyDays ? row.qualityReasons.join(', ') || 'DATA_ANOMALY' : null,
source: row.validDays > 0 ? 'ONEOS_API' : 'NONE', isDataSynced: row.validDays > 0,
isOnline: row.periodMileageKm != null && row.periodMileageKm > 0,
sourceProtocol: row.endDateSourceProtocol,
sourceCategory: categories.size > 1 ? 'MIXED' : categories.values().next().value || null,
dataTime: row.endDateDataTime, updatedAt: row.updatedAt, calculatedAt: null,
};
});
}
+11 -1
View File
@@ -3,6 +3,13 @@ export interface CachedVehicle {
plate: string;
vin: string;
dailyKm: number;
periodMileageKm?: number | null;
validDays?: number;
missingDays?: number;
anomalyDays?: number;
qualityStatus?: 'NORMAL' | 'PARTIAL_DATA' | 'NO_DATA' | 'DATA_ANOMALY';
qualityReasons?: string[];
sourceProtocols?: ('GB32960' | 'MQTT' | 'JT808')[];
mileageAnomaly?: string | null;
dailyMileage?: Record<string, number | null>;
totalKm: number | null;
@@ -64,6 +71,8 @@ export interface MonitoringStats {
totalAll: number;
vehicleCount: number;
yesterdayTotal: number;
validVehicleCount?: number;
totalMileageVehicleCount?: number;
}
/** /monitoring 完整响应 */
@@ -71,11 +80,12 @@ export interface MonitoringResponse {
vehicles: CachedVehicle[];
stats: MonitoringStats;
filters: MonitoringFilters;
rangeDailyTotals?: { date: string; totalKm: number }[];
rangeDailyTotals?: { date: string; totalKm: number | null }[];
dateRange?: { start: string; end: string };
total: number;
page: number;
totalPages: number;
queriedAt?: string;
updatedAt: string;
sourcePriority: ('instrument' | 'gps')[];
}
@@ -1,4 +1,7 @@
import { Hono } from 'hono';
import { fetchVehicleInfoByPlates } from './vehicle-info.js';
import { filterByPermission } from '../../auth/permissions.js';
import type { AuthUser } from '../../auth/types.js';
import { validDailyMileage } from './oneos-model.js';
import { fetchOneOsMileageDates } from './oneos-api.js';
import {
@@ -60,6 +63,12 @@ app.get('/:plate/recent', async (c) => {
}
try {
const user = c.get('user' as never) as AuthUser | undefined;
if (user) {
const info = await fetchVehicleInfoByPlates([plate]);
const visible = filterByPermission([...info.values()].map(v => ({ ...v, managerId: v.manager_id })), user);
if (!visible.length) return c.json({ error: '无权查看该车辆里程' }, 403);
}
const dates: string[] = [];
const dateCursor = new Date(start);
while (dateCursor <= end) {