Skip to content
Projects
Groups
Snippets
Help
This project
Loading...
Sign in / Register
Toggle navigation
P
pac
Overview
Overview
Details
Activity
Cycle Analytics
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Charts
Issues
0
Issues
0
List
Board
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Charts
Wiki
Wiki
Snippets
Snippets
Members
Collapse sidebar
Close sidebar
Activity
Graph
Charts
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
ai-tools
pac
Commits
c80c42f1
Commit
c80c42f1
authored
Oct 17, 2026
by
luoqi
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
fix(sync): normalize empty DW EMR entries and align lag cursor monitoring
parent
1cafc8ae
Hide whitespace changes
Inline
Side-by-side
Showing
9 changed files
with
114 additions
and
2 deletions
+114
-2
apps/pac-service/data/jvs-dw/assemblers/emr.yaml
+8
-0
apps/pac-service/src/modules/sync/assembler/assembler-engine.ts
+4
-0
apps/pac-service/src/modules/sync/assembler/assembler.schema.ts
+5
-0
apps/pac-service/src/modules/sync/assembler/empty-narrative.ts
+13
-0
apps/pac-service/src/queues/dw-lag-monitor.service.ts
+14
-2
apps/pac-service/tests/dw-lag-partial.spec.ts
+26
-0
apps/pac-service/tests/empty-narrative.spec.ts
+8
-0
apps/pac-service/tests/emr-normalization.spec.ts
+15
-0
docs/qa/2026-10-17-dw-emr-alert-fix.md
+21
-0
No files found.
apps/pac-service/data/jvs-dw/assemblers/emr.yaml
View file @
c80c42f1
...
@@ -48,3 +48,11 @@ join_arrays:
...
@@ -48,3 +48,11 @@ join_arrays:
foreign_key
:
emr_id
foreign_key
:
emr_id
same_source_row
:
true
same_source_row
:
true
field_mapping
:
{
toothPosition
:
tooth_position
,
message
:
treat_name
}
field_mapping
:
{
toothPosition
:
tooth_position
,
message
:
treat_name
}
# 旧牙位编辑器会留下空元素/仅牙位元素,无正文不等于整份病历无效。
# 非空元数据、未知字段和错误 message 类型仍拒绝;原始内容保留在 rawPayload。
empty_narrative_entries
:
examFindings
:
empty_metadata_keys
:
&empty_editor_keys
[
englishValTopRight
,
englishVal
,
englishValBotRight
,
englishValBotLeft
]
disposal
:
empty_metadata_keys
:
*empty_editor_keys
apps/pac-service/src/modules/sync/assembler/assembler-engine.ts
View file @
c80c42f1
No preview for this file type
apps/pac-service/src/modules/sync/assembler/assembler.schema.ts
View file @
c80c42f1
...
@@ -111,6 +111,11 @@ export const AssemblerConfigSchema = z.object({
...
@@ -111,6 +111,11 @@ export const AssemblerConfigSchema = z.object({
/// 主表字段 → canonical 字段映射(canonical key ← host field)
/// 主表字段 → canonical 字段映射(canonical key ← host field)
field_mapping
:
fieldMapping
,
field_mapping
:
fieldMapping
,
/// 显式兼容无正文的病历数组元素;仅白名单元数据为空时补 message:null。
empty_narrative_entries
:
z
.
record
(
z
.
string
(),
z
.
object
({
empty_metadata_keys
:
z
.
array
(
z
.
string
().
min
(
1
)),
})).
optional
(),
/// 主表字段的枚举映射(key 是 canonical 字段名)
/// 主表字段的枚举映射(key 是 canonical 字段名)
enum_mapping
:
z
.
record
(
z
.
string
(),
enumMappingForField
).
optional
(),
enum_mapping
:
z
.
record
(
z
.
string
(),
enumMappingForField
).
optional
(),
...
...
apps/pac-service/src/modules/sync/assembler/empty-narrative.ts
0 → 100644
View file @
c80c42f1
/** 宿主在 assembler 显式声明才启用。保留牙位,绝不把未知字段/非空编辑器内容吞成空正文。 */
export
function
normalizeEmptyNarrative
(
value
:
unknown
,
emptyMetadataKeys
:
string
[]):
unknown
{
if
(
!
Array
.
isArray
(
value
))
return
value
;
const
allowed
=
new
Set
([
'toothPosition'
,
...
emptyMetadataKeys
]);
return
value
.
map
(
entry
=>
{
if
(
!
entry
||
typeof
entry
!==
'object'
||
Array
.
isArray
(
entry
)
||
Object
.
prototype
.
hasOwnProperty
.
call
(
entry
,
'message'
))
return
entry
;
if
(
Object
.
keys
(
entry
).
some
(
key
=>
!
allowed
.
has
(
key
)))
return
entry
;
if
(
emptyMetadataKeys
.
some
(
key
=>
entry
[
key
]
!=
null
&&
entry
[
key
]
!==
''
&&
!
(
Array
.
isArray
(
entry
[
key
])
&&
entry
[
key
].
length
===
0
)))
return
entry
;
return
{
...
entry
,
message
:
null
};
});
}
apps/pac-service/src/queues/dw-lag-monitor.service.ts
View file @
c80c42f1
...
@@ -7,7 +7,7 @@ import { ColdImportService } from '../modules/sync/cold-import/cold-import.servi
...
@@ -7,7 +7,7 @@ import { ColdImportService } from '../modules/sync/cold-import/cold-import.servi
/**
/**
* DwLagMonitorService — 每小时检查 DW 增量数据滞后(W4 末)
* DwLagMonitorService — 每小时检查 DW 增量数据滞后(W4 末)
*
*
* 输入:sync_logs 中最近一条成功的 incremental_bundle 的 cursor_after JSON
* 输入:sync_logs 中最近一条成功
或部分成功
的 incremental_bundle 的 cursor_after JSON
* 例:{"fact_emr_treatment_out":"2026-05-26 15:30:00",...}
* 例:{"fact_emr_treatment_out":"2026-05-26 15:30:00",...}
*
*
* 计算:max(cursor_after.*) vs now() → diff hours
* 计算:max(cursor_after.*) vs now() → diff hours
...
@@ -62,7 +62,8 @@ export class DwLagMonitorService {
...
@@ -62,7 +62,8 @@ export class DwLagMonitorService {
where
:
{
where
:
{
hostId
:
host
.
id
,
hostId
:
host
.
id
,
resource
:
'incremental_bundle'
,
resource
:
'incremental_bundle'
,
status
:
'success'
,
// 与实际增量读取水位一致:partial 也已推进游标。
status
:
{
in
:
[
'success'
,
'partial'
]
},
cursorAfter
:
{
not
:
null
},
cursorAfter
:
{
not
:
null
},
},
},
orderBy
:
{
startedAt
:
'desc'
},
orderBy
:
{
startedAt
:
'desc'
},
...
@@ -71,6 +72,13 @@ export class DwLagMonitorService {
...
@@ -71,6 +72,13 @@ export class DwLagMonitorService {
this
.
logger
.
warn
(
`dw-lag: host=
${
host
.
name
}
还没跑过增量(无 cursor)— 检查 sync-incremental cron`
);
this
.
logger
.
warn
(
`dw-lag: host=
${
host
.
name
}
还没跑过增量(无 cursor)— 检查 sync-incremental cron`
);
continue
;
continue
;
}
}
if
(
last
.
status
===
'partial'
&&
last
.
failed
>
0
)
{
await
this
.
alerter
.
send
({
level
:
'warning'
,
title
:
'DW 同步部分失败'
,
body
:
`最近一轮有
${
last
.
failed
}
条处理失败;游标已推进不代表数据完整,请核查失败记录并定向补处理。`
,
context
:
{
host
:
host
.
name
,
syncLogId
:
last
.
id
,
failed
:
last
.
failed
,
startedAt
:
last
.
startedAt
.
toISOString
()
},
});
}
let
cursors
:
Record
<
string
,
string
>
;
let
cursors
:
Record
<
string
,
string
>
;
try
{
try
{
cursors
=
JSON
.
parse
(
last
.
cursorAfter
)
as
Record
<
string
,
string
>
;
cursors
=
JSON
.
parse
(
last
.
cursorAfter
)
as
Record
<
string
,
string
>
;
...
@@ -83,6 +91,10 @@ export class DwLagMonitorService {
...
@@ -83,6 +91,10 @@ export class DwLagMonitorService {
.
map
((
v
)
=>
new
Date
(
v
).
getTime
())
.
map
((
v
)
=>
new
Date
(
v
).
getTime
())
.
filter
((
t
)
=>
!
isNaN
(
t
)),
.
filter
((
t
)
=>
!
isNaN
(
t
)),
);
);
if
(
!
Number
.
isFinite
(
maxCursorTs
))
{
this
.
logger
.
warn
(
`dw-lag: host=
${
host
.
name
}
无有效游标时间,无法判断滞后`
);
continue
;
}
const
diffHours
=
(
now
-
maxCursorTs
)
/
3
_600_000
;
const
diffHours
=
(
now
-
maxCursorTs
)
/
3
_600_000
;
const
tag
=
const
tag
=
diffHours
>
errorH
?
'🔴'
:
diffHours
>
errorH
?
'🔴'
:
...
...
apps/pac-service/tests/dw-lag-partial.spec.ts
0 → 100644
View file @
c80c42f1
import
{
DwLagMonitorService
}
from
'../src/queues/dw-lag-monitor.service'
;
import
{
Logger
}
from
'@nestjs/common'
;
beforeAll
(()
=>
Logger
.
overrideLogger
(
false
));
afterAll
(()
=>
Logger
.
overrideLogger
([
'log'
,
'warn'
,
'error'
]));
async
function
check
(
age
:
number
,
status
=
'partial'
,
invalid
=
false
)
{
const
latest
=
{
id
:
'run'
,
status
,
failed
:
status
===
'partial'
?
3
:
0
,
startedAt
:
new
Date
(),
cursorAfter
:
JSON
.
stringify
({
emr
:
invalid
?
'invalid'
:
new
Date
(
Date
.
now
()
-
age
*
3600000
).
toISOString
()
})
};
const
old
=
{
...
latest
,
status
:
'success'
,
failed
:
0
,
cursorAfter
:
JSON
.
stringify
({
emr
:
new
Date
(
Date
.
now
()
-
25
*
3600000
).
toISOString
()
})
};
const
prisma
=
{
host
:
{
findMany
:
jest
.
fn
().
mockResolvedValue
([{
id
:
'host'
,
name
:
'jvs-dw'
}])
},
syncLog
:
{
findFirst
:
jest
.
fn
(
async
({
where
})
=>
where
.
status
===
'success'
?
old
:
latest
)
}
};
const
alert
=
{
send
:
jest
.
fn
().
mockResolvedValue
(
undefined
)
};
const
config
=
{
getHostOpsConfig
:
()
=>
({
hasSqlSource
:
true
,
pushLagAlert
:
false
})
};
await
new
DwLagMonitorService
(
prisma
as
any
,
alert
as
any
,
config
as
any
).
checkLag
();
return
{
prisma
,
alert
};
}
test
(
'新 partial 替代旧 success,报告部分失败而非假滞后'
,
async
()
=>
{
const
{
prisma
,
alert
}
=
await
check
(
3
);
expect
(
prisma
.
syncLog
.
findFirst
.
mock
.
calls
[
0
]
!
[
0
].
where
.
status
).
toEqual
({
in
:
[
'success'
,
'partial'
]
});
expect
(
alert
.
send
.
mock
.
calls
.
map
(
x
=>
x
[
0
].
title
)).
toEqual
([
'DW 同步部分失败'
]);
});
test
(
'partial 游标真的旧仍报滞后,失败不被掩盖'
,
async
()
=>
{
const
{
alert
}
=
await
check
(
30
);
expect
(
alert
.
send
.
mock
.
calls
.
map
(
x
=>
x
[
0
].
title
)).
toEqual
([
'DW 同步部分失败'
,
'DW 数据滞后(注意)'
]);
});
test
(
'success 正常不报警'
,
async
()
=>
expect
((
await
check
(
3
,
'success'
)).
alert
.
send
).
not
.
toHaveBeenCalled
());
test
(
'无有效时间不抛 RangeError'
,
async
()
=>
await
expect
(
check
(
3
,
'success'
,
true
)).
resolves
.
toBeDefined
());
apps/pac-service/tests/empty-narrative.spec.ts
0 → 100644
View file @
c80c42f1
import
{
normalizeEmptyNarrative
}
from
'../src/modules/sync/assembler/empty-narrative'
;
import
{
EmrNarrativeEntrySchema
}
from
'@pac/types'
;
test
.
each
([{
value
:
'不能丢的正文'
},
{
englishVal
:
[
'nonempty'
]
},
{
message
:
7
},
{
toothPosition
:
7
},
null
,
'text'
])(
'非空或未知结构不能变为空正文 %p'
,
entry
=>
{
const
[
result
]
=
normalizeEmptyNarrative
([
entry
],
[
'englishVal'
])
as
unknown
[];
expect
(
EmrNarrativeEntrySchema
.
safeParse
(
result
).
success
).
toBe
(
false
);
});
test
(
'字符串保持不变'
,
()
=>
expect
(
normalizeEmptyNarrative
(
'正常文字'
,
[])).
toBe
(
'正常文字'
));
apps/pac-service/tests/emr-normalization.spec.ts
View file @
c80c42f1
...
@@ -137,3 +137,18 @@ test('话术同时保留一般情况、辅助检查和未分类治疗原文,
...
@@ -137,3 +137,18 @@ test('话术同时保留一般情况、辅助检查和未分类治疗原文,
expect
(
text
).
toContain
(
'治疗计划:复查后评估'
);
expect
(
text
).
toContain
(
'治疗计划:复查后评估'
);
expect
(
text
).
not
.
toContain
(
'"message"'
);
expect
(
text
).
not
.
toContain
(
'"message"'
);
});
});
test
.
each
([
'examine'
,
'dispose'
])(
'5i5ya %s 兼容空旧编辑器、牙位和空对象,保留有效正文与原文'
,
field
=>
{
const
entries
=
[{
englishVal
:
[],
englishValTopRight
:
[],
englishValBotRight
:
[],
englishValBotLeft
:
[]
},
{
toothPosition
:
'16'
},
{},
{
toothPosition
:
'26'
,
message
:
'原文'
}];
const
raw
=
{
id
:
'E-empty'
,
patient_id
:
'P1'
,
organization_id
:
'C1'
,
brand
:
'瑞尔'
,
rq
:
'2026-10-17'
,
created_date
:
'2026-10-17 12:00:00'
,
updated_date
:
'2026-10-17 18:32:38'
,
[
field
]:
JSON
.
stringify
(
entries
)
};
const
saved
=
JSON
.
stringify
(
raw
);
const
{
canonical
,
rawSource
}
=
replay
(
'jvs-dw'
,
raw
,
'fact_emr_treatment_out'
);
expect
(
canonical
[
field
===
'examine'
?
'examFindings'
:
'disposal'
]).
toEqual
([
{
message
:
null
},
{
toothPosition
:
'16'
,
message
:
null
},
{
message
:
null
},
{
toothPosition
:
'26'
,
message
:
'原文'
},
]);
expect
(
JSON
.
stringify
(
rawSource
)).
toBe
(
saved
);
});
docs/qa/2026-10-17-dw-emr-alert-fix.md
0 → 100644
View file @
c80c42f1
# 5i5ya 空病历元素兼容与游标告警修复
## 问题与边界
增量任务允许 partial 并继续使用其 cursorAfter,但滞后监控只读取 success,导致最近任务仅有少量坏记录时,误用更早成功任务的游标。监控现在与读取水位一致,并单独报告部分失败;游标推进不等于数据完整。召回计算耗时与并发策略不在本次修改范围。
5i5ya 的 examine / dispose 存在缺少 message 的空元素:仅牙位,或旧编辑器的空数组。仅在 jvs-dw assembler 显式启用兼容,将这些元素的正文补为 null,保留牙位及完整原始记录。未知字段、非空编辑器内容、错误类型仍拒绝,不放宽通用病历协议。
## 本地验证
-
5 个相关测试套件,40 项通过;共享包构建、Prisma 生成、服务类型检查和构建通过。
-
使用源端以下 3 条实际记录,在临时隔离数据库通过真实 assembler → transaction → parser → fact writer 落库,均成功;再次重放全部 unchanged,没有新增事实版本。临时数据库已删除。
-
未启动画像或召回计算,未改动宿主游标。远端补处理前保留对应交易与事实快照。
| 源病历 ID | 兼容字段 |
| --- | --- |
| a0edf43927ab4a508e141b29c1b00071 | examine |
| 2210d9c4664147cbaaf75455718c77bd | dispose |
| 7e4f2f9f2d3343e79017cef97f6f3e06 | dispose |
发布顺序:独立 feature 合入 test,测试环境定向重放验证后,同一 feature 合入 main;不将 test 合回 main。历史 partial 日志保留原样,因此补处理成功后,部分失败告警仍可能指向旧任务,直到下一轮正常增量完成。
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment