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
01b9ae78
Commit
01b9ae78
authored
Oct 20, 2026
by
luoqi
Browse files
Options
Browse Files
Download
Plain Diff
Merge image-first Meyer ingestion into test
parents
6c3f5f8e
1ebcfd66
Pipeline
#3835
failed in 0 seconds
Changes
12
Pipelines
1
Hide whitespace changes
Inline
Side-by-side
Showing
12 changed files
with
266 additions
and
55 deletions
+266
-55
apps/pac-service/src/cli/meyer-imaging.cli.ts
+6
-4
apps/pac-service/src/integrations/meyer/README.md
+16
-13
apps/pac-service/src/integrations/meyer/collector.ts
+39
-15
apps/pac-service/src/integrations/meyer/contracts.ts
+5
-3
apps/pac-service/src/integrations/meyer/mapper.ts
+36
-0
apps/pac-service/src/integrations/meyer/pac-sink.ts
+17
-2
apps/pac-service/src/integrations/meyer/scan.ts
+2
-2
apps/pac-service/src/integrations/meyer/scheduler.ts
+1
-1
apps/pac-service/tests/meyer-imaging-postgres.spec.ts
+66
-3
apps/pac-service/tests/meyer-imaging.spec.ts
+60
-6
apps/pac-service/tests/meyer-resume.spec.ts
+4
-4
apps/pac-service/tests/meyer-scheduler.spec.ts
+14
-2
No files found.
apps/pac-service/src/cli/meyer-imaging.cli.ts
View file @
01b9ae78
...
@@ -74,7 +74,7 @@ export async function runMeyerCli(argv: string[]): Promise<number> {
...
@@ -74,7 +74,7 @@ export async function runMeyerCli(argv: string[]): Promise<number> {
const
matches
:
Array
<
{
key
:
string
;
saved
:
SavedBundle
}
>
=
[];
const
matches
:
Array
<
{
key
:
string
;
saved
:
SavedBundle
}
>
=
[];
for
(
const
scope
of
config
.
scopes
)
{
for
(
const
scope
of
config
.
scopes
)
{
const
key
=
imageKey
(
scope
,
args
.
imageFiles
);
const
key
=
imageKey
(
scope
,
args
.
imageFiles
);
const
saved
=
await
store
.
read
<
SavedBundle
>
(
`bundle-
${
key
}
`
);
const
saved
=
await
store
.
read
<
SavedBundle
>
(
`bundle-
${
key
}
`
)
??
await
store
.
read
<
SavedBundle
>
(
`image-
${
key
}
`
)
;
if
(
saved
)
matches
.
push
({
key
,
saved
});
if
(
saved
)
matches
.
push
({
key
,
saved
});
}
}
if
(
matches
.
length
!==
1
)
throw
new
Error
(
'requires exactly one collected image in configured scope'
);
if
(
matches
.
length
!==
1
)
throw
new
Error
(
'requires exactly one collected image in configured scope'
);
...
@@ -99,14 +99,16 @@ export async function runMeyerCli(argv: string[]): Promise<number> {
...
@@ -99,14 +99,16 @@ export async function runMeyerCli(argv: string[]): Promise<number> {
if
(
args
.
tick
)
{
if
(
args
.
tick
)
{
const
status
=
await
runMeyerTick
(
config
,
c
=>
client
.
forWindow
(
c
),
store
,
codes
,
sink
!
);
const
status
=
await
runMeyerTick
(
config
,
c
=>
client
.
forWindow
(
c
),
store
,
codes
,
sink
!
);
console
.
log
(
JSON
.
stringify
({
mode
:
'tick'
,
pending
:
status
.
pending
,
retryAttempted
:
status
.
retryAttempted
,
console
.
log
(
JSON
.
stringify
({
mode
:
'tick'
,
pending
:
status
.
pending
,
retryAttempted
:
status
.
retryAttempted
,
retryFailed
:
status
.
retryFailed
,
scanned
:
status
.
scan
?.
examined
??
0
,
scanComplete
:
status
.
scan
?.
complete
??
false
,
scanFailed
:
status
.
scan
?.
failed
??
0
}));
retryFailed
:
status
.
retryFailed
,
scanned
:
status
.
scan
?.
examined
??
0
,
return
status
.
retryFailed
||
status
.
scan
?.
failed
?
1
:
status
.
pending
?
2
:
0
;
scanComplete
:
status
.
scan
?.
complete
??
false
,
scanFailed
:
status
.
scan
?.
failed
??
0
,
imagesApplied
:
status
.
scan
?.
imagesApplied
??
0
,
analysisFailed
:
status
.
scan
?.
analysisFailed
??
0
}));
return
status
.
retryFailed
||
status
.
scan
?.
failed
||
status
.
scan
?.
analysisFailed
?
1
:
status
.
pending
?
2
:
0
;
}
}
const
result
=
await
collectMeyer
(
config
,
client
,
store
,
codes
,
sink
);
const
result
=
await
collectMeyer
(
config
,
client
,
store
,
codes
,
sink
);
// 控制台只输出数量;原文、患者标识和逐条错误留在私有状态目录。
// 控制台只输出数量;原文、患者标识和逐条错误留在私有状态目录。
const
{
errors
:
_errors
,
...
counts
}
=
result
;
const
{
errors
:
_errors
,
...
counts
}
=
result
;
console
.
log
(
JSON
.
stringify
({
mode
:
args
.
apply
?
'apply'
:
'preview'
,
...
counts
}));
console
.
log
(
JSON
.
stringify
({
mode
:
args
.
apply
?
'apply'
:
'preview'
,
...
counts
}));
return
result
.
failed
?
1
:
result
.
deferred
?
2
:
0
;
return
result
.
failed
||
result
.
analysisFailed
?
1
:
result
.
deferred
?
2
:
0
;
}
finally
{
}
finally
{
try
{
await
app
?.
close
();
}
finally
{
await
release
();
}
try
{
await
app
?.
close
();
}
finally
{
await
release
();
}
}
}
...
...
apps/pac-service/src/integrations/meyer/README.md
View file @
01b9ae78
...
@@ -19,9 +19,9 @@
...
@@ -19,9 +19,9 @@
独立 CLI → 美亚只读接口 → 私有目录留底与标准化 → PAC 已有事实解析/写入内核。
独立 CLI → 美亚只读接口 → 私有目录留底与标准化 → PAC 已有事实解析/写入内核。
-
只补 FridaySaaS 宿主现有患者的影像
诊断信号
,不新增宿主、患者或归属规则。
-
只补 FridaySaaS 宿主现有患者的影像
元数据与影像分析
,不新增宿主、患者或归属规则。
-
独立进程、凭据、诊所范围和状态目录,不注册到 AppModule,不修改 Push/Pull、主摄入游标、画像算法或召回规则。
-
独立进程、凭据、诊所范围和状态目录,不注册到 AppModule,不修改 Push/Pull、主摄入游标、画像算法或召回规则。
-
本期不做对账:
只接入首次读到的有效牙位结论;已成功接入的影像不复查分析
、不处理修改/删除/撤回/改绑。发现已留底影像改绑则报错,等待后续对账能力。
-
本期不做对账:
影像先入库,首次有效分析晚到后补充;分析成功后不再复查
、不处理修改/删除/撤回/改绑。发现已留底影像改绑则报错,等待后续对账能力。
-
默认固定窗口;
`--tick`
执行一轮独立增量任务,
`--watch`
每轮完成后按配置间隔(默认 15 分钟)再运行。逐诊所分页、串行限速,使用独立水位和待处理队列。
-
默认固定窗口;
`--tick`
执行一轮独立增量任务,
`--watch`
每轮完成后按配置间隔(默认 15 分钟)再运行。逐诊所分页、串行限速,使用独立水位和待处理队列。
-
不直接触发画像或召回计算;新事实进入现有事务/事实水位,由已启用的 PAC 既有 stale-scan 等任务消费。该任务未启用的本地环境不会自动重算。
-
不直接触发画像或召回计算;新事实进入现有事务/事实水位,由已启用的 PAC 既有 stale-scan 等任务消费。该任务未启用的本地环境不会自动重算。
...
@@ -44,7 +44,7 @@
...
@@ -44,7 +44,7 @@
美亚没有提供分析完成状态、分析时间和版本。本期不伪造
`completed/analyzedAt`
,不绕过既有 image_analysis 的完整快照要求;将已观察到的牙位结论转换为现有
`image_study`
、
`diagnosis`
,使用原有解析器和 FactWriter。
**这是检查时刻的影像发现,不是医生诊断或美亚分析完成时刻。**
原文、观察时间与
`timeBasis=studyTime`
保存在适配器文件和事务 rawPayload 中。
美亚没有提供分析完成状态、分析时间和版本。本期不伪造
`completed/analyzedAt`
,不绕过既有 image_analysis 的完整快照要求;将已观察到的牙位结论转换为现有
`image_study`
、
`diagnosis`
,使用原有解析器和 FactWriter。
**这是检查时刻的影像发现,不是医生诊断或美亚分析完成时刻。**
原文、观察时间与
`timeBasis=studyTime`
保存在适配器文件和事务 rawPayload 中。
没有检查时间、没有入库时间、空 teeth 或未绑定患者只留底待处理
。综合诊断/描述不拆分为新诊断。临时 URL 不写入长期影像地址;原片沿用宿主患者级入口。本期不新增 UI;若宿主使用单张影像模板,须先确认其能识别此来源的影像标识。
已绑定患者且有入库时间即可保存影像,不以分析是否存在为前提。无分析时持久化待处理,之后重试补充到同一影像;分析接口异常也不影响已入库影像或后续影像扫描。没有入库时间或未绑定患者仍留底待处理,不伪造时间或患者归属。缺检查时间时可以保存影像内分析,但不生成带推测时间的诊断
。综合诊断/描述不拆分为新诊断。临时 URL 不写入长期影像地址;原片沿用宿主患者级入口。本期不新增 UI;若宿主使用单张影像模板,须先确认其能识别此来源的影像标识。
## 运行
## 运行
...
@@ -60,9 +60,9 @@ pnpm meyer-imaging -- --config=/absolute/meyer.json --state-dir=/absolute/privat
...
@@ -60,9 +60,9 @@ pnpm meyer-imaging -- --config=/absolute/meyer.json --state-dir=/absolute/privat
pnpm meyer-imaging
--
--config
=
/absolute/meyer.json
--state-dir
=
/absolute/private/meyer-test
--apply
pnpm meyer-imaging
--
--config
=
/absolute/meyer.json
--state-dir
=
/absolute/private/meyer-test
--apply
```
```
退出码:0=所扫范围无失败或待处理;1=请求/校验/写入失败;2=仍有待处理记录。0 不代表美亚全部历史数据或所有分析均已同步。终端只输出计数;细项见
`last-run.json`
,原始响应见
`observed-*`
,
标准字段及应用状态
见
`bundle-*`
。
退出码:0=所扫范围无失败或待处理;1=请求/校验/写入失败;2=仍有待处理记录。0 不代表美亚全部历史数据或所有分析均已同步。终端只输出计数;细项见
`last-run.json`
,原始响应见
`observed-*`
,
影像标准字段及应用状态见
`image-*`
,分析快照
见
`bundle-*`
。
成功标记在所有事实落库后才写入。部分失败重跑同窗,使用已留底快照恢复;事务身份稳定,重复写入不新增事实
。状态目录有进程锁;明确属于本机已退出进程的锁可自动回收。存活进程、其他机器、未知 owner 或回收操作本身中断的锁不自动抢占。更换范围映射、应用或 PAC 数据库须使用独立目录;单独扩大时间窗可复用原目录。
影像和分析分别在对应事实成功落库后标记。影像元数据与分析使用两个稳定事件身份,影像业务 ID 保持一致;分析到达后由原有 FactWriter 更新同一影像的版本,并关联牙位诊断。部分失败使用已留底快照恢复;重复执行不增加重复事实,重放元数据不会清空分析
。状态目录有进程锁;明确属于本机已退出进程的锁可自动回收。存活进程、其他机器、未知 owner 或回收操作本身中断的锁不自动抢占。更换范围映射、应用或 PAC 数据库须使用独立目录;单独扩大时间窗可复用原目录。
不要把补充记录交给 Friday 原表 YAML 的通用 reparse;本路线以自己的留底快照重试。首次 preview 留底后,apply 使用同一快照,不重新解释为最新结果。已应用文件不可直接改写或删除来模拟对账。
不要把补充记录交给 Friday 原表 YAML 的通用 reparse;本路线以自己的留底快照重试。首次 preview 留底后,apply 使用同一快照,不重新解释为最新结果。已应用文件不可直接改写或删除来模拟对账。
...
@@ -71,7 +71,7 @@ pnpm meyer-imaging -- --config=/absolute/meyer.json --state-dir=/absolute/privat
...
@@ -71,7 +71,7 @@ pnpm meyer-imaging -- --config=/absolute/meyer.json --state-dir=/absolute/privat
-
已用测试患者完成真实只读联调:PDF 三项身份参数可用,分析可落库,文件 ID 可换取 OSS 签名地址。详见 QA 记录。
-
已用测试患者完成真实只读联调:PDF 三项身份参数可用,分析可落库,文件 ID 可换取 OSS 签名地址。详见 QA 记录。
-
无更新游标、完整快照分页和删除标记,无法承诺完整生命周期同步;分页数量变化、重复/缺页会报错,重跑固定窗口。只在空响应和缺字段时延后,不能把它们当作阴性结论。
-
无更新游标、完整快照分页和删除标记,无法承诺完整生命周期同步;分页数量变化、重复/缺页会报错,重跑固定窗口。只在空响应和缺字段时延后,不能把它们当作阴性结论。
-
宿主若以后推送同一批美亚影像,必须先统一来源标识和接入责任,避免与本路线重复。首期应只开明确授权的试点诊所。
-
宿主若以后推送同一批美亚影像,必须先统一来源标识和接入责任,避免与本路线重复。首期应只开明确授权的试点诊所。
-
本期不部署。开发回归使用本地 HTTP 与独立 PostgreSQL 数据库;真实联调只读源端、仅写隔离的本地验证
库。
-
测试环境曾运行固定窗口联调;这不等于已启用持续任务。启用前应核实实际进程、启动配置、持久状态及扫描水位。开发回归使用本地 HTTP 与独立 PostgreSQL 数据
库。
## 独立增量与待分析重试
## 独立增量与待分析重试
...
@@ -84,14 +84,17 @@ pnpm meyer-imaging -- --config=/absolute/meyer.json --state-dir=/absolute/privat
...
@@ -84,14 +84,17 @@ pnpm meyer-imaging -- --config=/absolute/meyer.json --state-dir=/absolute/privat
pnpm meyer-imaging
--
--config
=
/absolute/meyer.json
--state-dir
=
/absolute/private/meyer-test
--apply
--watch
pnpm meyer-imaging
--
--config
=
/absolute/meyer.json
--state-dir
=
/absolute/private/meyer-test
--apply
--watch
```
```
首次从配置 timeStart 开始,后续使用私有 cursor。tick/watch 的结束时间由当前时间决定(固定窗口 timeEnd 仅用于默认单次采集),每轮最多推进 1 小时,落后当前时刻至少 1 分钟,前后窗口重叠 1 分钟
。时间窗口不是分析更新时间,不保证捕获任意历史修订。
诊所范围是授权边界,不应因定时运行而扩大。首次从配置 timeStart 开始,后续使用私有 cursor。tick/watch 的结束时间由当前时间决定(固定窗口 timeEnd 仅用于默认单次采集),每轮最多推进 1 小时,落后当前时刻至少 1 分钟,前后窗口重叠 1 分钟。时间窗口按影像入库时间划分,不要求影像在窗口内已经分析;晚到分析由 pending 跨窗口追踪
。时间窗口不是分析更新时间,不保证捕获任意历史修订。
-
新影像没有分析时先持久化到 pending/,之后允许扫描水位前进。分页、身份校验或落库失败时不推进。
单次固定窗口用于联调或明确范围的历史补采,不会自动变成定时任务。持续采集须由外部守护真正运行
`--watch`
,或定时运行
`--tick`
;仅保存配置不会启用。历史跨度较大时应先完成明确范围的补采,随后使用同一目录从初始时间逐窗确认水位;不要把一次窄窗成功当成历史完整,也不要手改 cursor 跳过尚未扫描的区间。
-
新影像先经过原有标准解析/事实写入流程入库。没有分析、分析请求失败或分析写入失败时保留 pending/,允许影像扫描水位前进;分页、身份校验或影像落库失败时不推进。
-
`imagesApplied`
记录本轮新写入影像阶段数,
`applied`
记录完成分析阶段数;
`failed`
是阻止影像扫描的失败,
`analysisFailed`
单独记录分析请求/写入异常,两类均通过 CLI 失败退出码反映。空分析计入 deferred,不制造分析完成或阴性结论。
-
每轮先处理最早到期的最多 10 个(可配置)待处理项,按 1 分钟起指数退避、最多 1 小时;不会因重试次数多而自动丢弃。
-
每轮先处理最早到期的最多 10 个(可配置)待处理项,按 1 分钟起指数退避、最多 1 小时;不会因重试次数多而自动丢弃。
-
重试通过原影像入库时间前后 1 秒的窄窗重新定位 imageId,再获取分析。不会扫描整个诊所历史。身份改绑、影像不可见均保留待处理并报错,不自动撤销或迁移事实。
-
重试通过原影像入库时间前后 1 秒的窄窗重新定位 imageId,再获取分析。不会扫描整个诊所历史。身份改绑、影像不可见均保留待处理并报错,不自动撤销或迁移事实。
-
因接口没有按 imageId 直接查询元数据的能力,缺少入库时间的条目不能自动定位,须人工处理;这类情况会有 retryFailed。
-
因接口没有按 imageId 直接查询元数据的能力,缺少入库时间的条目不能自动定位,须人工处理;这类情况会有 retryFailed。
-
scheduler-status.json 提供扫描结果、水位、待处理数、重试失败数。watch 遇网络、美亚或数据库异常会记录失败计数并退避,默认从 15 分钟增长到最多 1 小时;依赖恢复后继续,不丢弃失败数据。配置错误、损坏状态或归属冲突需要修复后才能恢复。SIGINT/SIGTERM 会等当前一轮收尾再停止。
-
scheduler-status.json 提供扫描结果、水位、待处理数、重试失败数。watch 遇网络、美亚或数据库异常会记录失败计数并退避,默认从 15 分钟增长到最多 1 小时;依赖恢复后继续,不丢弃失败数据。配置错误、损坏状态或归属冲突需要修复后才能恢复。SIGINT/SIGTERM 会等当前一轮收尾再停止。
-
已成功
影像不重新查询分析
。这是首次接入及晚到分析重试,不是历史对账;删除、撤回和成功后的修改仍不覆盖。
-
已成功
分析的影像不再重新查询分析;只有影像阶段完成时仍会重试
。这是首次接入及晚到分析重试,不是历史对账;删除、撤回和成功后的修改仍不覆盖。
-
状态目录必须持久化。先前仅有 deferred-
*
的旧目录没有完整队列,升级后先按原窗口重跑一次再启动 tick/watch。不要用曾做远期手工采集的目录直接推断历史已覆盖。
-
状态目录必须持久化。先前仅有 deferred-
*
的旧目录没有完整队列,升级后先按原窗口重跑一次再启动 tick/watch。不要用曾做远期手工采集的目录直接推断历史已覆盖。
## 原图地址按需刷新
## 原图地址按需刷新
...
@@ -101,7 +104,7 @@ pnpm meyer-imaging -- --config=/absolute/meyer.json --state-dir=/absolute/privat
...
@@ -101,7 +104,7 @@ pnpm meyer-imaging -- --config=/absolute/meyer.json --state-dir=/absolute/privat
pnpm meyer-imaging
--
--config
=
/absolute/meyer.json
--state-dir
=
/absolute/private/meyer-test
--image-files
=
<美亚imageId>
pnpm meyer-imaging
--
--config
=
/absolute/meyer.json
--state-dir
=
/absolute/private/meyer-test
--image-files
=
<美亚imageId>
```
```
从
bundle
中取原图 imageOssId,并使用对应诊所调用美亚 images/urls。签名 URL 写入私有 access-
*
.json,终端只显示 found、expiresAt 和文件键,不输出签名。每次执行重新获取,不能把过期地址当作长期业务数据。当前命令只刷新原图,不批量获取 CT 切片。
从
分析 bundle 或影像 image 快照
中取原图 imageOssId,并使用对应诊所调用美亚 images/urls。签名 URL 写入私有 access-
*
.json,终端只显示 found、expiresAt 和文件键,不输出签名。每次执行重新获取,不能把过期地址当作长期业务数据。当前命令只刷新原图,不批量获取 CT 切片。
这是运维 CLI 能力,不是公开下载接口;未来若接前端,必须先按宿主患者权限授权,再调用访问解析。美亚 URL 接口只校验诊所文件归属,不能替代患者级权限。
这是运维 CLI 能力,不是公开下载接口;未来若接前端,必须先按宿主患者权限授权,再调用访问解析。美亚 URL 接口只校验诊所文件归属,不能替代患者级权限。
...
@@ -133,11 +136,11 @@ scan-progress 保存固定窗口、诊所、页码、页内位置和当前页原
...
@@ -133,11 +136,11 @@ scan-progress 保存固定窗口、诊所、页码、页内位置和当前页原
| 改绑、字段不合法、权限失效 | 不跳过或篡改;记录状态,需修复根因后恢复 |
| 改绑、字段不合法、权限失效 | 不跳过或篡改;记录状态,需修复根因后恢复 |
| 状态文件损坏、磁盘满 | 不推进水位;修复磁盘/恢复备份后继续 |
| 状态文件损坏、磁盘满 | 不推进水位;修复磁盘/恢复备份后继续 |
提供
`meyer-imaging.service.example`
作为 Linux 进程守护模板
,本次没有安装或启用
。进程自身无法在被 SIGKILL/OOM 杀死后自我重启,必须有外部守护。watch 还有单轮看门狗:数据库请求等卡住超过默认 10 分钟时退出,避免旧轮尚未停止就并发启动新轮;重启后使用持久断点恢复。
提供
`meyer-imaging.service.example`
作为 Linux 进程守护模板
;是否安装、启用应以目标环境的实际状态为准
。进程自身无法在被 SIGKILL/OOM 杀死后自我重启,必须有外部守护。watch 还有单轮看门狗:数据库请求等卡住超过默认 10 分钟时退出,避免旧轮尚未停止就并发启动新轮;重启后使用持久断点恢复。
### 存储分工
### 存储分工
-
**独立状态目录**
:identity、数据库目标指纹、cursor、scan-progress、scan-ledger、pending、bundle、observed、scheduler-status、access 文件;本期不新增数据库采集表。状态目录须放持久磁盘、限制权限、备份。只支持单机单写者,不用于多机器共享抢任务。
-
**独立状态目录**
:identity、数据库目标指纹、cursor、scan-progress、scan-ledger、pending、
image、
bundle、observed、scheduler-status、access 文件;本期不新增数据库采集表。状态目录须放持久磁盘、限制权限、备份。只支持单机单写者,不用于多机器共享抢任务。
-
**PAC 共用表**
:标准化事件写 patient_transactions,标准影像/诊断写 patient_facts,按 Friday 宿主、租户、品牌与患者隔离;patients 只查找,不创建或移动。原始美亚依据还保留在事务 rawPayload。
-
**PAC 共用表**
:标准化事件写 patient_transactions,标准影像/诊断写 patient_facts,按 Friday 宿主、租户、品牌与患者隔离;patients 只查找,不创建或移动。原始美亚依据还保留在事务 rawPayload。
-
**原片**
:仍在美亚/OSS,本地只留文件 ID 和业务证据,不复制文件。
-
**原片**
:仍在美亚/OSS,本地只留文件 ID 和业务证据,不复制文件。
...
@@ -151,4 +154,4 @@ scan-progress 保存固定窗口、诊所、页码、页内位置和当前页原
...
@@ -151,4 +154,4 @@ scan-progress 保存固定窗口、诊所、页码、页内位置和当前页原
5i5ya 可使用同一结构,但当前映射只把分析关联到病历,没有可靠的单张影像标识,因此不填影像
`analysis`
。缺失/null 不等于源端没有分析数据。
5i5ya 可使用同一结构,但当前映射只把分析关联到病历,没有可靠的单张影像标识,因此不填影像
`analysis`
。缺失/null 不等于源端没有分析数据。
本次不补写历史:旧 bundle 缺失
`analysis`
时继续按原快照重试,已应用的快照继续跳过;旧事实不自动增加字段。丢失状态后对已入库影像生成不同快照,仍按原有哈希检查拒绝覆盖
。不得删除状态目录来强制升级历史。
兼容旧 v1 bundle:按原快照重试,已应用的快照继续跳过;旧事实不自动增加字段。新 v2 快照分为影像与分析阶段,前者没有
`analysis`
,后者补充同一影像。已有完整分析的影像遇到元数据重放时保留分析;不同分析快照仍按原有哈希检查拒绝覆盖,后续修订/撤回不在本次范围内
。不得删除状态目录来强制升级历史。
apps/pac-service/src/integrations/meyer/collector.ts
View file @
01b9ae78
import
{
join
}
from
'node:path'
;
import
{
join
}
from
'node:path'
;
import
{
ImageSchema
,
SymptomSchema
,
hash
,
imageKey
,
sourceTime
,
import
{
ImageSchema
,
SymptomSchema
,
hash
,
imageKey
,
sourceTime
,
type
MeyerConfig
,
type
MeyerSink
,
type
MeyerSource
,
type
MeyerImage
,
type
MeyerScope
}
from
'./contracts'
;
type
MeyerConfig
,
type
MeyerSink
,
type
MeyerSource
,
type
MeyerImage
,
type
MeyerScope
}
from
'./contracts'
;
import
{
mapMeyer
}
from
'./mapper'
;
import
{
mapMeyer
Image
,
mapMeyerAnalysis
}
from
'./mapper'
;
import
{
MeyerStore
,
type
SavedBundle
}
from
'./store'
;
import
{
MeyerStore
,
type
SavedBundle
}
from
'./store'
;
export
interface
PendingImage
{
export
interface
PendingImage
{
...
@@ -9,7 +9,7 @@ export interface PendingImage {
...
@@ -9,7 +9,7 @@ export interface PendingImage {
}
}
export
interface
CollectResult
{
export
interface
CollectResult
{
pages
:
number
;
seen
:
number
;
collected
:
number
;
applied
:
number
;
alreadyApplied
:
number
;
pages
:
number
;
seen
:
number
;
collected
:
number
;
applied
:
number
;
alreadyApplied
:
number
;
deferred
:
number
;
failed
:
number
;
errors
:
Array
<
{
key
:
string
;
reason
:
string
}
>
;
deferred
:
number
;
failed
:
number
;
imagesApplied
:
number
;
analysisFailed
:
number
;
errors
:
Array
<
{
key
:
string
;
reason
:
string
}
>
;
}
}
/** 固定时间窗单次运行;不推进主流程游标,不删除、不改绑、不复查已完成影像。 */
/** 固定时间窗单次运行;不推进主流程游标,不删除、不改绑、不复查已完成影像。 */
...
@@ -17,7 +17,7 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
...
@@ -17,7 +17,7 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
codes
:
Record
<
string
,
string
>
,
sink
?:
MeyerSink
,
respectRetryAt
=
false
):
Promise
<
CollectResult
>
{
codes
:
Record
<
string
,
string
>
,
sink
?:
MeyerSink
,
respectRetryAt
=
false
):
Promise
<
CollectResult
>
{
const
pending
=
new
MeyerStore
(
join
(
store
.
directory
,
'pending'
));
const
pending
=
new
MeyerStore
(
join
(
store
.
directory
,
'pending'
));
const
result
:
CollectResult
=
{
pages
:
0
,
seen
:
0
,
collected
:
0
,
applied
:
0
,
const
result
:
CollectResult
=
{
pages
:
0
,
seen
:
0
,
collected
:
0
,
applied
:
0
,
alreadyApplied
:
0
,
deferred
:
0
,
failed
:
0
,
errors
:
[]
};
alreadyApplied
:
0
,
deferred
:
0
,
failed
:
0
,
imagesApplied
:
0
,
analysisFailed
:
0
,
errors
:
[]
};
for
(
const
scope
of
config
.
scopes
)
{
for
(
const
scope
of
config
.
scopes
)
{
const
seen
=
new
Set
<
string
>
();
const
seen
=
new
Set
<
string
>
();
let
total
:
number
|
undefined
;
let
total
:
number
|
undefined
;
...
@@ -38,6 +38,7 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
...
@@ -38,6 +38,7 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
let
key
=
`page-
${
page
}
-row-
${
result
.
seen
}
`
;
let
key
=
`page-
${
page
}
-row-
${
result
.
seen
}
`
;
let
retry
:
PendingImage
|
undefined
;
let
retry
:
PendingImage
|
undefined
;
let
completed
=
false
;
let
completed
=
false
;
let
imageReady
=
false
;
try
{
try
{
const
parsed
=
ImageSchema
.
safeParse
(
raw
);
const
parsed
=
ImageSchema
.
safeParse
(
raw
);
if
(
!
parsed
.
success
)
throw
new
Error
(
'invalid image contract'
);
if
(
!
parsed
.
success
)
throw
new
Error
(
'invalid image contract'
);
...
@@ -49,19 +50,42 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
...
@@ -49,19 +50,42 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
if
(
image
.
imageCreateTime
&&
(
sourceTime
(
image
.
imageCreateTime
)
<
sourceTime
(
config
.
timeStart
)
||
if
(
image
.
imageCreateTime
&&
(
sourceTime
(
image
.
imageCreateTime
)
<
sourceTime
(
config
.
timeStart
)
||
sourceTime
(
image
.
imageCreateTime
)
>
sourceTime
(
config
.
timeEnd
)))
throw
new
Error
(
'image outside requested window'
);
sourceTime
(
image
.
imageCreateTime
)
>
sourceTime
(
config
.
timeEnd
)))
throw
new
Error
(
'image outside requested window'
);
const
previousRetry
=
await
pending
.
read
<
PendingImage
>
(
key
);
const
previousRetry
=
await
pending
.
read
<
PendingImage
>
(
key
);
if
(
respectRetryAt
&&
previousRetry
&&
Date
.
parse
(
previousRetry
.
nextRetryAt
)
>
Date
.
now
())
{
result
.
deferred
++
;
continue
;
}
retry
=
{
scope
,
image
,
attempts
:
(
previousRetry
?.
attempts
??
0
)
+
1
,
nextRetryAt
:
''
};
retry
=
{
scope
,
image
,
attempts
:
(
previousRetry
?.
attempts
??
0
)
+
1
,
nextRetryAt
:
''
};
const
saved
=
await
store
.
read
<
SavedBundle
>
(
`bundle-
${
key
}
`
);
const
saved
=
await
store
.
read
<
SavedBundle
>
(
`bundle-
${
key
}
`
);
if
(
saved
&&
(
hash
(
saved
.
bundle
.
scope
)
!==
hash
(
scope
)
||
saved
.
bundle
.
namespace
!==
config
.
namespace
||
const
metadata
=
await
store
.
read
<
SavedBundle
>
(
`image-
${
key
}
`
);
saved
.
bundle
.
raw
.
image
.
imageId
!==
image
.
imageId
||
saved
.
bundle
.
version
!==
1
))
{
for
(
const
retained
of
[
saved
,
metadata
])
{
throw
new
Error
(
'saved bundle scope mismatch'
);
if
(
!
retained
)
continue
;
}
if
(
hash
(
retained
.
bundle
.
scope
)
!==
hash
(
scope
)
||
retained
.
bundle
.
namespace
!==
config
.
namespace
||
if
(
saved
&&
saved
.
bundle
.
image
.
patientExternalId
!==
image
.
patientId
)
{
retained
.
bundle
.
raw
.
image
.
imageId
!==
image
.
imageId
||
throw
new
Error
(
'image ownership changed; reconciliation is not supported'
);
!
[
1
,
2
].
includes
(
retained
.
bundle
.
version
)
||
(
retained
===
metadata
&&
(
retained
.
bundle
.
version
!==
2
||
retained
.
bundle
.
phase
!==
'image'
))
||
(
retained
===
saved
&&
retained
.
bundle
.
version
===
2
&&
retained
.
bundle
.
phase
!==
'analysis'
))
{
throw
new
Error
(
'saved bundle scope mismatch'
);
}
if
(
retained
.
bundle
.
image
.
patientExternalId
!==
image
.
patientId
)
{
throw
new
Error
(
'image ownership changed; reconciliation is not supported'
);
}
}
}
if
(
saved
?.
applied
)
{
completed
=
true
;
result
.
alreadyApplied
++
;
continue
;
}
if
(
saved
?.
applied
)
{
completed
=
true
;
result
.
alreadyApplied
++
;
continue
;
}
// v1 重试仍使用原完整快照。新采集必须先成功保存影像,再请求分析。
if
(
saved
?.
bundle
.
version
!==
1
)
{
const
imageBundle
=
metadata
?.
bundle
??
mapMeyerImage
(
config
,
scope
,
image
,
new
Date
().
toISOString
());
if
(
!
imageBundle
)
{
await
store
.
write
(
`deferred-
${
key
}
`
,
{
reason
:
'unbound patient or missing image creation time'
,
image
});
result
.
deferred
++
;
continue
;
}
if
(
!
metadata
)
await
store
.
write
(
`image-
${
key
}
`
,
{
bundle
:
imageBundle
,
applied
:
false
}
satisfies
SavedBundle
);
if
(
sink
&&
!
metadata
?.
applied
)
{
await
sink
.
apply
(
imageBundle
);
await
store
.
write
(
`image-
${
key
}
`
,
{
bundle
:
imageBundle
,
applied
:
true
}
satisfies
SavedBundle
);
result
.
imagesApplied
++
;
}
imageReady
=
true
;
if
(
respectRetryAt
&&
previousRetry
&&
Date
.
parse
(
previousRetry
.
nextRetryAt
)
>
Date
.
now
())
{
// 不因扫描重遇同一行延长原重试时间。
retry
=
undefined
;
result
.
deferred
++
;
continue
;
}
}
let
bundle
=
saved
?.
bundle
;
let
bundle
=
saved
?.
bundle
;
if
(
!
bundle
)
{
if
(
!
bundle
)
{
const
observedAt
=
new
Date
().
toISOString
();
const
observedAt
=
new
Date
().
toISOString
();
...
@@ -74,10 +98,10 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
...
@@ -74,10 +98,10 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
await
store
.
write
(
`observed-
${
key
}
`
,
{
observedAt
,
image
,
symptom
:
rawSymptom
});
await
store
.
write
(
`observed-
${
key
}
`
,
{
observedAt
,
image
,
symptom
:
rawSymptom
});
const
symptom
=
SymptomSchema
.
safeParse
(
rawSymptom
);
const
symptom
=
SymptomSchema
.
safeParse
(
rawSymptom
);
if
(
!
symptom
.
success
)
throw
new
Error
(
'invalid symptom contract'
);
if
(
!
symptom
.
success
)
throw
new
Error
(
'invalid symptom contract'
);
bundle
=
mapMeyer
(
config
,
scope
,
image
,
symptom
.
data
,
codes
,
observedAt
)
??
undefined
;
bundle
=
mapMeyer
Analysis
(
config
,
scope
,
image
,
symptom
.
data
,
codes
,
observedAt
)
??
undefined
;
if
(
!
bundle
)
{
if
(
!
bundle
)
{
await
store
.
write
(
`deferred-
${
key
}
`
,
{
observedAt
,
await
store
.
write
(
`deferred-
${
key
}
`
,
{
observedAt
,
reason
:
'
no tooth findings or missing image/study tim
e'
});
reason
:
'
analysis not yet availabl
e'
});
result
.
deferred
++
;
continue
;
result
.
deferred
++
;
continue
;
}
}
// 写入前固定完整快照;失败重试使用同一快照,不混入后来变化的远端结果。
// 写入前固定完整快照;失败重试使用同一快照,不混入后来变化的远端结果。
...
@@ -91,7 +115,7 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
...
@@ -91,7 +115,7 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
}
}
completed
=
!!
sink
;
completed
=
!!
sink
;
}
catch
(
error
)
{
}
catch
(
error
)
{
result
.
failed
++
;
if
(
imageReady
)
result
.
analysisFailed
++
;
else
result
.
failed
++
;
result
.
errors
.
push
({
key
,
reason
:
error
instanceof
Error
?
error
.
message
:
'collection failed'
});
result
.
errors
.
push
({
key
,
reason
:
error
instanceof
Error
?
error
.
message
:
'collection failed'
});
}
finally
{
}
finally
{
if
(
retry
)
{
if
(
retry
)
{
...
...
apps/pac-service/src/integrations/meyer/contracts.ts
View file @
01b9ae78
...
@@ -77,15 +77,17 @@ export const SymptomSchema = z.object({
...
@@ -77,15 +77,17 @@ export const SymptomSchema = z.object({
export
type
MeyerSymptom
=
z
.
infer
<
typeof
SymptomSchema
>
;
export
type
MeyerSymptom
=
z
.
infer
<
typeof
SymptomSchema
>
;
export
interface
MeyerBundle
{
export
interface
MeyerBundle
{
version
:
1
;
version
:
1
|
2
;
/** v1 是历史完整快照;v2 分影像和分析两阶段,不改写旧留底。 */
phase
?:
'image'
|
'analysis'
;
namespace
:
string
;
namespace
:
string
;
scope
:
MeyerScope
;
scope
:
MeyerScope
;
observedAt
:
string
;
observedAt
:
string
;
// 分析状态/分析时刻未提供。诊断时间是影像检查时刻,绝不以采集时刻代替。
// 分析状态/分析时刻未提供。诊断时间是影像检查时刻,绝不以采集时刻代替。
timeBasis
:
'studyTime'
;
timeBasis
:
'studyTime'
|
'imageCreateTime'
;
image
:
ImageStudyCanonical
;
image
:
ImageStudyCanonical
;
diagnoses
:
DiagnosisCanonical
[];
diagnoses
:
DiagnosisCanonical
[];
raw
:
{
image
:
MeyerImage
;
symptom
:
MeyerSymptom
};
raw
:
{
image
:
MeyerImage
;
symptom
:
MeyerSymptom
|
null
};
}
}
export
interface
MeyerSink
{
apply
(
bundle
:
MeyerBundle
):
Promise
<
void
>
}
export
interface
MeyerSink
{
apply
(
bundle
:
MeyerBundle
):
Promise
<
void
>
}
export
interface
MeyerSource
{
export
interface
MeyerSource
{
...
...
apps/pac-service/src/integrations/meyer/mapper.ts
View file @
01b9ae78
...
@@ -52,3 +52,39 @@ export function mapMeyer(config: MeyerConfig, scope: MeyerScope, image: MeyerIma
...
@@ -52,3 +52,39 @@ export function mapMeyer(config: MeyerConfig, scope: MeyerScope, image: MeyerIma
}
}),
}
}),
diagnoses
,
raw
:
{
image
,
symptom
}
};
diagnoses
,
raw
:
{
image
,
symptom
}
};
}
}
/** 影像元数据独立接入;没有分析或检查时间也不阻止保存影像。 */
export
function
mapMeyerImage
(
config
:
MeyerConfig
,
scope
:
MeyerScope
,
image
:
MeyerImage
,
observedAt
:
string
):
MeyerBundle
|
null
{
if
(
image
.
clinicId
!==
scope
.
clinicId
)
throw
new
Error
(
'Meyer clinic differs from configured scope'
);
if
(
!
image
.
patientId
||
!
image
.
imageCreateTime
)
return
null
;
const
uploadedAt
=
sourceTime
(
image
.
imageCreateTime
);
if
(
uploadedAt
>
observedAt
)
throw
new
Error
(
'Meyer image has future time'
);
return
{
version
:
2
,
phase
:
'image'
,
namespace
:
config
.
namespace
,
scope
,
observedAt
,
timeBasis
:
'imageCreateTime'
,
diagnoses
:
[],
raw
:
{
image
,
symptom
:
null
},
image
:
ImageStudyCanonicalSchema
.
parse
({
externalId
:
`meyer:
${
config
.
namespace
}
:
${
encodeURIComponent
(
image
.
imageId
)}
`
,
patientExternalId
:
image
.
patientId
,
clinicId
:
scope
.
clinicId
,
uploadedAt
,
modality
:
[
'pa'
,
'bw'
,
'pano'
,
'cbct'
,
'intraoral_photo'
,
'scan'
,
'other'
].
includes
(
image
.
modality
??
''
)
?
image
.
modality
:
'other'
,
fileType
:
image
.
imageExt
??
null
,
})
};
}
/** 晚到分析补到同一张影像;缺检查时间时只保存分析,不制造诊断时间。 */
export
function
mapMeyerAnalysis
(
config
:
MeyerConfig
,
scope
:
MeyerScope
,
image
:
MeyerImage
,
symptom
:
MeyerSymptom
,
codes
:
Record
<
string
,
string
>
,
observedAt
:
string
):
MeyerBundle
|
null
{
const
metadata
=
mapMeyerImage
(
config
,
scope
,
image
,
observedAt
);
if
(
!
metadata
||
(
!
symptom
.
teeth
.
length
&&
!
symptom
.
diagnosis
.
trim
()
&&
!
symptom
.
description
.
trim
()))
return
null
;
// 有检查时刻和牙位时沿用原来的诊断映射,保持 v1 已入库事务哈希兼容。
const
legacy
=
mapMeyer
(
config
,
scope
,
image
,
symptom
,
codes
,
observedAt
);
if
(
legacy
)
return
{
...
legacy
,
version
:
2
,
phase
:
'analysis'
};
if
(
image
.
studyTime
&&
sourceTime
(
image
.
studyTime
)
>
observedAt
)
throw
new
Error
(
'Meyer image has future time'
);
const
findings
=
[...
new
Map
(
symptom
.
teeth
.
map
(
t
=>
[
hash
([
t
.
position
,
t
.
content
]),
t
])).
entries
()]
.
sort
(([
a
],
[
b
])
=>
a
.
localeCompare
(
b
));
return
{
...
metadata
,
phase
:
'analysis'
,
timeBasis
:
image
.
studyTime
?
'studyTime'
:
'imageCreateTime'
,
image
:
ImageStudyCanonicalSchema
.
parse
({
...
metadata
.
image
,
analysis
:
{
source
:
'meyer'
,
summary
:
symptom
.
diagnosis
||
null
,
description
:
symptom
.
description
||
null
,
analyzedAt
:
null
,
findings
:
findings
.
map
(([,
t
])
=>
({
name
:
t
.
content
,
toothPosition
:
t
.
position
,
code
:
resolveMeyerDiagnosisCode
(
t
.
content
,
codes
)
})),
}
}),
raw
:
{
image
,
symptom
}
};
}
apps/pac-service/src/integrations/meyer/pac-sink.ts
View file @
01b9ae78
...
@@ -17,7 +17,11 @@ export class MeyerPacSink implements MeyerSink {
...
@@ -17,7 +17,11 @@ export class MeyerPacSink implements MeyerSink {
const
{
tenantId
,
sourceUnit
,
clinicId
}
=
bundle
.
scope
;
const
{
tenantId
,
sourceUnit
,
clinicId
}
=
bundle
.
scope
;
const
image
=
ImageStudyCanonicalSchema
.
parse
(
bundle
.
image
);
const
image
=
ImageStudyCanonicalSchema
.
parse
(
bundle
.
image
);
const
diagnoses
=
bundle
.
diagnoses
.
map
(
d
=>
DiagnosisCanonicalSchema
.
parse
(
d
));
const
diagnoses
=
bundle
.
diagnoses
.
map
(
d
=>
DiagnosisCanonicalSchema
.
parse
(
d
));
if
(
!
diagnoses
.
length
||
diagnoses
.
some
(
d
=>
d
.
patientExternalId
!==
image
.
patientExternalId
||
if
(
!
[
1
,
2
].
includes
(
bundle
.
version
)
||
(
bundle
.
version
===
1
&&
!
diagnoses
.
length
)
||
(
bundle
.
version
===
2
&&
(
!
[
'image'
,
'analysis'
].
includes
(
bundle
.
phase
??
''
)
||
(
bundle
.
phase
===
'image'
&&
(
image
.
analysis
!==
undefined
||
diagnoses
.
length
>
0
))
||
(
bundle
.
phase
===
'analysis'
&&
!
image
.
analysis
)))
||
diagnoses
.
some
(
d
=>
d
.
patientExternalId
!==
image
.
patientExternalId
||
d
.
clinicId
!==
clinicId
||
d
.
sourceImageExternalId
!==
image
.
externalId
)
||
image
.
clinicId
!==
clinicId
)
{
d
.
clinicId
!==
clinicId
||
d
.
sourceImageExternalId
!==
image
.
externalId
)
||
image
.
clinicId
!==
clinicId
)
{
throw
new
Error
(
'canonical bundle ownership mismatch'
);
throw
new
Error
(
'canonical bundle ownership mismatch'
);
}
}
...
@@ -32,6 +36,16 @@ export class MeyerPacSink implements MeyerSink {
...
@@ -32,6 +36,16 @@ export class MeyerPacSink implements MeyerSink {
...
diagnoses
.
map
(
row
=>
({
resource
:
'diagnosis'
,
row
,
action
:
Action
.
DIAGNOSIS_RECORDED
,
...
diagnoses
.
map
(
row
=>
({
resource
:
'diagnosis'
,
row
,
action
:
Action
.
DIAGNOSIS_RECORDED
,
subjectType
:
SubjectType
.
DIAGNOSIS
,
timeField
:
'occurredAt'
}))];
subjectType
:
SubjectType
.
DIAGNOSIS
,
timeField
:
'occurredAt'
}))];
for
(
const
item
of
rows
)
{
for
(
const
item
of
rows
)
{
// 元数据重试不能清掉已补充的分析;患者归属始终按完整身份校验。
if
(
bundle
.
version
===
2
&&
bundle
.
phase
===
'image'
)
{
const
current
=
await
this
.
prisma
.
patientFact
.
findFirst
({
where
:
{
hostId
:
host
.
id
,
tenantId
,
sourceUnit
,
subjectId
:
`image_record:
${
image
.
externalId
}
`
,
},
orderBy
:
{
version
:
'desc'
}
});
if
(
current
&&
(
current
.
patientId
!==
patient
.
id
||
current
.
clinicId
!==
clinicId
))
{
throw
new
Error
(
'existing supplemental snapshot differs; image ownership changed'
);
}
if
((
current
?.
content
as
{
analysis
?:
unknown
}
|
undefined
)?.
analysis
)
continue
;
}
const
canonicalHash
=
hash
(
item
.
resource
===
'image_study'
?
{
image
,
diagnoses
}
:
item
.
row
);
const
canonicalHash
=
hash
(
item
.
resource
===
'image_study'
?
{
image
,
diagnoses
}
:
item
.
row
);
const
rawRow
=
{
...
item
.
row
,
_meyer
:
{
namespace
:
bundle
.
namespace
,
const
rawRow
=
{
...
item
.
row
,
_meyer
:
{
namespace
:
bundle
.
namespace
,
timeBasis
:
bundle
.
timeBasis
,
observedAt
:
bundle
.
observedAt
,
sourceUnit
,
canonicalHash
,
timeBasis
:
bundle
.
timeBasis
,
observedAt
:
bundle
.
observedAt
,
sourceUnit
,
canonicalHash
,
...
@@ -39,7 +53,8 @@ export class MeyerPacSink implements MeyerSink {
...
@@ -39,7 +53,8 @@ export class MeyerPacSink implements MeyerSink {
const
data
=
this
.
synthesizer
.
synthesize
({
rawRow
,
canonicalRow
:
item
.
row
,
const
data
=
this
.
synthesizer
.
synthesize
({
rawRow
,
canonicalRow
:
item
.
row
,
emits
:
{
action
:
item
.
action
,
subjectType
:
item
.
subjectType
,
occurredAtField
:
item
.
timeField
,
contentVersion
:
false
},
emits
:
{
action
:
item
.
action
,
subjectType
:
item
.
subjectType
,
occurredAtField
:
item
.
timeField
,
contentVersion
:
false
},
resource
:
item
.
resource
,
hostId
:
host
.
id
,
tenantId
,
sourceUnit
,
resource
:
item
.
resource
,
hostId
:
host
.
id
,
tenantId
,
sourceUnit
,
patientIdResolver
:
()
=>
patient
.
id
,
sourceContext
:
`supplement:meyer:
${
bundle
.
namespace
}
`
});
patientIdResolver
:
()
=>
patient
.
id
,
sourceContext
:
`supplement:meyer:
${
bundle
.
namespace
}
`
+
(
bundle
.
version
===
2
&&
bundle
.
phase
===
'image'
?
':image'
:
''
)
});
if
(
!
data
)
throw
new
Error
(
'canonical transaction rejected'
);
if
(
!
data
)
throw
new
Error
(
'canonical transaction rejected'
);
// 固定业务身份;重复提交只恢复未完成事实,不把不同内容静默当作重试。
// 固定业务身份;重复提交只恢复未完成事实,不把不同内容静默当作重试。
await
this
.
prisma
.
patientTransaction
.
createMany
({
data
:
[
data
],
skipDuplicates
:
true
});
await
this
.
prisma
.
patientTransaction
.
createMany
({
data
:
[
data
],
skipDuplicates
:
true
});
...
...
apps/pac-service/src/integrations/meyer/scan.ts
View file @
01b9ae78
...
@@ -33,7 +33,7 @@ export async function scanWindow(config: MeyerConfig, sourceFor: (c: MeyerConfig
...
@@ -33,7 +33,7 @@ export async function scanWindow(config: MeyerConfig, sourceFor: (c: MeyerConfig
const
window
=
{
...
config
,
timeStart
:
start
,
timeEnd
:
end
,
pageSize
:
progress
.
pageSize
};
const
window
=
{
...
config
,
timeStart
:
start
,
timeEnd
:
end
,
pageSize
:
progress
.
pageSize
};
const
source
=
sourceFor
(
window
);
const
source
=
sourceFor
(
window
);
const
result
:
CollectResult
&
{
complete
:
boolean
;
examined
:
number
}
=
{
pages
:
0
,
seen
:
0
,
collected
:
0
,
const
result
:
CollectResult
&
{
complete
:
boolean
;
examined
:
number
}
=
{
pages
:
0
,
seen
:
0
,
collected
:
0
,
applied
:
0
,
alreadyApplied
:
0
,
deferred
:
0
,
failed
:
0
,
errors
:
[],
complete
:
false
,
examined
:
0
};
applied
:
0
,
alreadyApplied
:
0
,
deferred
:
0
,
failed
:
0
,
imagesApplied
:
0
,
analysisFailed
:
0
,
errors
:
[],
complete
:
false
,
examined
:
0
};
const
limit
=
config
.
maxListPagesPerTick
;
const
limit
=
config
.
maxListPagesPerTick
;
let
validatedScope
=
-
1
;
let
validatedScope
=
-
1
;
const
fetchPage
=
async
(
scope
:
MeyerScope
,
page
:
number
)
=>
{
const
fetchPage
=
async
(
scope
:
MeyerScope
,
page
:
number
)
=>
{
...
@@ -99,7 +99,7 @@ export async function scanWindow(config: MeyerConfig, sourceFor: (c: MeyerConfig
...
@@ -99,7 +99,7 @@ export async function scanWindow(config: MeyerConfig, sourceFor: (c: MeyerConfig
list
:
async
()
=>
({
total
:
1
,
page
:
1
,
size
:
window
.
pageSize
,
images
:
[
image
]
}),
list
:
async
()
=>
({
total
:
1
,
page
:
1
,
size
:
window
.
pageSize
,
images
:
[
image
]
}),
symptom
:
im
=>
source
.
symptom
(
im
),
symptom
:
im
=>
source
.
symptom
(
im
),
},
store
,
codes
,
sink
,
true
);
},
store
,
codes
,
sink
,
true
);
for
(
const
key
of
[
'seen'
,
'collected'
,
'applied'
,
'alreadyApplied'
,
'deferred'
,
'failed'
]
as
const
)
result
[
key
]
+=
one
[
key
];
for
(
const
key
of
[
'seen'
,
'collected'
,
'applied'
,
'alreadyApplied'
,
'deferred'
,
'failed'
,
'imagesApplied'
,
'analysisFailed'
]
as
const
)
result
[
key
]
+=
one
[
key
];
result
.
errors
.
push
(...
one
.
errors
);
result
.
errors
.
push
(...
one
.
errors
);
if
(
one
.
failed
)
return
result
;
// 不跳过未知失败;重试从本行恢复。
if
(
one
.
failed
)
return
result
;
// 不跳过未知失败;重试从本行恢复。
progress
.
row
++
;
await
store
.
write
(
'scan-progress'
,
progress
);
progress
.
row
++
;
await
store
.
write
(
'scan-progress'
,
progress
);
...
...
apps/pac-service/src/integrations/meyer/scheduler.ts
View file @
01b9ae78
...
@@ -52,7 +52,7 @@ export async function runMeyerTick(config: MeyerConfig, sourceFor: (c: MeyerConf
...
@@ -52,7 +52,7 @@ export async function runMeyerTick(config: MeyerConfig, sourceFor: (c: MeyerConf
list
:
async
()
=>
({
total
:
1
,
page
:
1
,
size
:
config
.
pageSize
,
images
:
[
found
]
}),
list
:
async
()
=>
({
total
:
1
,
page
:
1
,
size
:
config
.
pageSize
,
images
:
[
found
]
}),
symptom
:
image
=>
client
.
symptom
(
image
),
symptom
:
image
=>
client
.
symptom
(
image
),
},
store
,
codes
,
sink
);
},
store
,
codes
,
sink
);
retryFailed
+=
result
.
failed
;
retryFailed
+=
result
.
failed
+
result
.
analysisFailed
;
}
catch
{
}
catch
{
retryFailed
++
;
retryFailed
++
;
await
pending
.
write
(
key
,
{
...
value
,
attempts
:
value
.
attempts
+
1
,
await
pending
.
write
(
key
,
{
...
value
,
attempts
:
value
.
attempts
+
1
,
...
...
apps/pac-service/tests/meyer-imaging-postgres.spec.ts
View file @
01b9ae78
...
@@ -11,6 +11,11 @@ import { PrismaService } from '../src/prisma/prisma.service';
...
@@ -11,6 +11,11 @@ import { PrismaService } from '../src/prisma/prisma.service';
import
{
MeyerCliModule
}
from
'../src/integrations/meyer/meyer-cli.module'
;
import
{
MeyerCliModule
}
from
'../src/integrations/meyer/meyer-cli.module'
;
import
{
MeyerPacSink
}
from
'../src/integrations/meyer/pac-sink'
;
import
{
MeyerPacSink
}
from
'../src/integrations/meyer/pac-sink'
;
import
{
ParserPipeline
}
from
'../src/modules/sync/pipeline/parser-pipeline.service'
;
import
{
ParserPipeline
}
from
'../src/modules/sync/pipeline/parser-pipeline.service'
;
import
{
collectMeyer
}
from
'../src/integrations/meyer/collector'
;
import
{
MeyerStore
}
from
'../src/integrations/meyer/store'
;
import
{
mapMeyerImage
,
mapMeyerAnalysis
}
from
'../src/integrations/meyer/mapper'
;
import
{
imageKey
}
from
'../src/integrations/meyer/contracts'
;
import
{
codes
}
from
'./meyer-imaging-fixtures'
;
import
{
meyerBundle
}
from
'./meyer-imaging-fixtures'
;
import
{
meyerBundle
}
from
'./meyer-imaging-fixtures'
;
import
{
meyerConfig
,
meyerImage
,
meyerSymptom
}
from
'./meyer-imaging-fixtures'
;
import
{
meyerConfig
,
meyerImage
,
meyerSymptom
}
from
'./meyer-imaging-fixtures'
;
import
{
runMeyerCli
}
from
'../src/cli/meyer-imaging.cli'
;
import
{
runMeyerCli
}
from
'../src/cli/meyer-imaging.cli'
;
...
@@ -111,6 +116,64 @@ describePg('Meyer | isolated PostgreSQL + actual minimal CLI module', () => {
...
@@ -111,6 +116,64 @@ describePg('Meyer | isolated PostgreSQL + actual minimal CLI module', () => {
await
expect
(
sink
.
apply
(
meyerBundle
())).
rejects
.
toThrow
(
'snapshot differs'
);
await
expect
(
sink
.
apply
(
meyerBundle
())).
rejects
.
toThrow
(
'snapshot differs'
);
expect
(
await
prisma
.
patientFact
.
count
({
where
:
{
hostId
}
})).
toBe
(
3
);
expect
(
await
prisma
.
patientFact
.
count
({
where
:
{
hostId
}
})).
toBe
(
3
);
});
});
test
(
'image-only fact survives analysis outage; late analysis enriches same subject and retries never downgrade'
,
async
()
=>
{
const
directory
=
await
mkdtemp
(
join
(
tmpdir
(),
'meyer-image-first-pg-'
));
const
config
=
meyerConfig
(),
image
=
meyerImage
(),
scope
=
config
.
scopes
[
0
]
!
;
const
store
=
new
MeyerStore
(
directory
);
let
ready
=
false
;
const
source
=
{
list
:
async
()
=>
({
total
:
1
,
page
:
1
,
size
:
100
,
images
:
[
image
]
}),
symptom
:
async
()
=>
{
if
(
!
ready
)
throw
Error
(
'analysis outage'
);
return
meyerSymptom
();
}
};
try
{
const
first
=
await
collectMeyer
(
config
,
source
,
store
,
codes
,
sink
);
expect
(
first
).
toMatchObject
({
imagesApplied
:
1
,
analysisFailed
:
1
,
failed
:
0
});
const
imageOnly
=
await
prisma
.
patientFact
.
findFirstOrThrow
({
where
:
{
hostId
,
patientId
}
});
expect
(
imageOnly
).
toMatchObject
({
type
:
FactType
.
IMAGE_RECORD
,
version
:
1
,
status
:
'active'
});
expect
(
imageOnly
.
content
).
not
.
toHaveProperty
(
'analysis'
);
expect
(
await
prisma
.
patientTransaction
.
count
({
where
:
{
hostId
}
})).
toBe
(
1
);
ready
=
true
;
expect
((
await
collectMeyer
(
config
,
source
,
store
,
codes
,
sink
)).
applied
).
toBe
(
1
);
const
versions
=
await
prisma
.
patientFact
.
findMany
({
where
:
{
hostId
,
patientId
,
type
:
FactType
.
IMAGE_RECORD
},
orderBy
:
{
version
:
'asc'
}
});
expect
(
versions
.
map
(
f
=>
[
f
.
subjectId
,
f
.
version
,
f
.
status
])).
toEqual
([
[
imageOnly
.
subjectId
,
1
,
'superseded'
],
[
imageOnly
.
subjectId
,
2
,
'active'
]]);
expect
(
versions
[
1
]
!
.
content
).
toMatchObject
({
analysis
:
{
source
:
'meyer'
,
findings
:
expect
.
any
(
Array
)
}
});
const
diagnoses
=
await
prisma
.
patientFact
.
findMany
({
where
:
{
hostId
,
patientId
,
type
:
FactType
.
DIAGNOSIS_RECORD
}
});
expect
(
diagnoses
).
toHaveLength
(
2
);
expect
(
diagnoses
.
every
(
f
=>
(
f
.
content
as
any
).
source_image_external_id
===
meyerBundle
().
image
.
externalId
)).
toBe
(
true
);
// 即使重放旧影像元数据,已完成分析也不能被清空。
await
sink
.
apply
(
mapMeyerImage
(
config
,
scope
,
image
,
'2026-10-17T00:00:00.000Z'
)
!
);
await
collectMeyer
(
config
,
source
,
store
,
codes
,
sink
);
expect
(
await
prisma
.
patientFact
.
count
({
where
:
{
hostId
}
})).
toBe
(
4
);
expect
(
await
prisma
.
patientTransaction
.
count
({
where
:
{
hostId
}
})).
toBe
(
4
);
const
current
=
await
prisma
.
patientFact
.
findFirstOrThrow
({
where
:
{
hostId
,
patientId
,
subjectId
:
imageOnly
.
subjectId
,
status
:
'active'
}
});
expect
(
current
.
version
).
toBe
(
2
);
expect
(
current
.
content
).
toHaveProperty
(
'analysis'
);
const
key
=
imageKey
(
scope
,
image
.
imageId
);
expect
(
await
store
.
read
(
`image-
${
key
}
`
)).
toMatchObject
({
applied
:
true
});
expect
(
await
new
MeyerStore
(
join
(
directory
,
'pending'
)).
keys
()).
toEqual
([]);
}
finally
{
await
rm
(
directory
,
{
recursive
:
true
,
force
:
true
});
}
});
test
(
'analysis without study time is kept on image without invented dated diagnosis'
,
async
()
=>
{
const
c
=
meyerConfig
(),
image
=
{
...
meyerImage
(),
studyTime
:
null
},
observedAt
=
'2026-10-17T00:00:00.000Z'
;
const
metadata
=
mapMeyerImage
(
c
,
c
.
scopes
[
0
]
!
,
image
,
observedAt
)
!
;
const
analysis
=
mapMeyerAnalysis
(
c
,
c
.
scopes
[
0
]
!
,
image
,
meyerSymptom
(),
codes
,
observedAt
)
!
;
await
sink
.
apply
(
metadata
);
await
sink
.
apply
(
analysis
);
await
sink
.
apply
(
analysis
);
expect
(
await
prisma
.
patientFact
.
count
({
where
:
{
hostId
,
type
:
FactType
.
DIAGNOSIS_RECORD
}
})).
toBe
(
0
);
const
current
=
await
prisma
.
patientFact
.
findFirstOrThrow
({
where
:
{
hostId
,
status
:
'active'
}
});
expect
(
current
.
content
).
toMatchObject
({
analysis
:
{
findings
:
expect
.
any
(
Array
),
analyzedAt
:
null
}
});
expect
(
current
.
occurredAt
?.
toISOString
()).
toBe
(
metadata
.
image
.
uploadedAt
);
});
test
(
'v2 rescan of existing legacy analysis is idempotent; changed snapshots still fail closed'
,
async
()
=>
{
const
c
=
meyerConfig
(),
observedAt
=
'2026-10-17T00:00:00.000Z'
;
await
sink
.
apply
(
meyerBundle
());
await
sink
.
apply
(
mapMeyerImage
(
c
,
c
.
scopes
[
0
]
!
,
meyerImage
(),
observedAt
)
!
);
await
sink
.
apply
(
mapMeyerAnalysis
(
c
,
c
.
scopes
[
0
]
!
,
meyerImage
(),
meyerSymptom
(),
codes
,
observedAt
)
!
);
expect
(
await
prisma
.
patientFact
.
count
({
where
:
{
hostId
}
})).
toBe
(
3
);
expect
(
await
prisma
.
patientTransaction
.
count
({
where
:
{
hostId
}
})).
toBe
(
3
);
const
changed
=
mapMeyerAnalysis
(
c
,
c
.
scopes
[
0
]
!
,
meyerImage
(),
{
...
meyerSymptom
(),
diagnosis
:
'changed'
},
codes
,
observedAt
)
!
;
await
expect
(
sink
.
apply
(
changed
)).
rejects
.
toThrow
(
'snapshot differs'
);
expect
(
await
prisma
.
patientFact
.
count
({
where
:
{
hostId
}
})).
toBe
(
3
);
});
test
(
'same external patient/image IDs stay isolated across Friday brands; other host untouched'
,
async
()
=>
{
test
(
'same external patient/image IDs stay isolated across Friday brands; other host untouched'
,
async
()
=>
{
const
a
=
meyerBundle
();
const
b
=
meyerBundle
();
b
.
scope
.
sourceUnit
=
'brand-b'
;
const
a
=
meyerBundle
();
const
b
=
meyerBundle
();
b
.
scope
.
sourceUnit
=
'brand-b'
;
await
sink
.
apply
(
a
);
await
sink
.
apply
(
b
);
await
sink
.
apply
(
a
);
await
sink
.
apply
(
b
);
...
@@ -180,8 +243,8 @@ describePg('Meyer | isolated PostgreSQL + actual minimal CLI module', () => {
...
@@ -180,8 +243,8 @@ describePg('Meyer | isolated PostgreSQL + actual minimal CLI module', () => {
const
args
=
[
`--config=
${
join
(
directory
,
'config.json'
)}
`
,
`--state-dir=
${
join
(
directory
,
'state'
)}
`
,
'--apply'
];
const
args
=
[
`--config=
${
join
(
directory
,
'config.json'
)}
`
,
`--state-dir=
${
join
(
directory
,
'state'
)}
`
,
'--apply'
];
expect
(
await
runMeyerCli
([...
args
,
'--tick'
])).
toBe
(
0
);
expect
(
await
runMeyerCli
([...
args
,
'--tick'
])).
toBe
(
0
);
expect
(
await
runMeyerCli
([...
args
,
'--tick'
])).
toBe
(
0
);
expect
(
await
runMeyerCli
([...
args
,
'--tick'
])).
toBe
(
0
);
expect
(
await
prisma
.
patientFact
.
count
({
where
:
{
hostId
,
patientId
}
})).
toBe
(
3
);
expect
(
await
prisma
.
patientFact
.
count
({
where
:
{
hostId
,
patientId
}
})).
toBe
(
4
);
expect
(
await
prisma
.
patientTransaction
.
count
({
where
:
{
hostId
}
})).
toBe
(
4
);
expect
(
await
prisma
.
patientTransaction
.
count
({
where
:
{
hostId
}
})).
toBe
(
5
);
await
prisma
.
host
.
update
({
where
:
{
id
:
hostId
},
data
:
{
await
prisma
.
host
.
update
({
where
:
{
id
:
hostId
},
data
:
{
appSecretHash
:
hashSecret
(
'local-meyer-open'
),
openFactsReadEnabled
:
true
,
appSecretHash
:
hashSecret
(
'local-meyer-open'
),
openFactsReadEnabled
:
true
,
clinicNames
:
{
'clinic-a'
:
'本地影像验收诊所'
},
clinicNames
:
{
'clinic-a'
:
'本地影像验收诊所'
},
...
@@ -236,7 +299,7 @@ describePg('Meyer | isolated PostgreSQL + actual minimal CLI module', () => {
...
@@ -236,7 +299,7 @@ describePg('Meyer | isolated PostgreSQL + actual minimal CLI module', () => {
await
hostRecord
(
otherPatientId
);
await
hostRecord
(
otherPatientId
);
await
expect
(
runMeyerCli
([...
args
,
'--tick'
])).
rejects
.
toThrow
(
'conflicting'
);
await
expect
(
runMeyerCli
([...
args
,
'--tick'
])).
rejects
.
toThrow
(
'conflicting'
);
expect
(
await
prisma
.
patientFact
.
count
({
where
:
{
hostId
,
patientId
}
})).
toBe
(
3
);
expect
(
await
prisma
.
patientFact
.
count
({
where
:
{
hostId
,
patientId
}
})).
toBe
(
4
);
}
finally
{
}
finally
{
await
open
?.
close
();
await
open
?.
close
();
output
.
mockRestore
();
output
.
mockRestore
();
...
...
apps/pac-service/tests/meyer-imaging.spec.ts
View file @
01b9ae78
...
@@ -7,7 +7,7 @@ import { ConfigSchema, ImageSchema, SymptomSchema, imageKey, sourceTime,
...
@@ -7,7 +7,7 @@ import { ConfigSchema, ImageSchema, SymptomSchema, imageKey, sourceTime,
type
MeyerBundle
,
type
MeyerConfig
,
type
MeyerSource
}
from
'../src/integrations/meyer/contracts'
;
type
MeyerBundle
,
type
MeyerConfig
,
type
MeyerSource
}
from
'../src/integrations/meyer/contracts'
;
import
{
MeyerClient
}
from
'../src/integrations/meyer/client'
;
import
{
MeyerClient
}
from
'../src/integrations/meyer/client'
;
import
{
collectMeyer
}
from
'../src/integrations/meyer/collector'
;
import
{
collectMeyer
}
from
'../src/integrations/meyer/collector'
;
import
{
loadDiagnosisCodes
,
mapMeyer
}
from
'../src/integrations/meyer/mapper'
;
import
{
loadDiagnosisCodes
,
mapMeyer
,
mapMeyerImage
,
mapMeyerAnalysis
}
from
'../src/integrations/meyer/mapper'
;
import
{
MeyerStore
,
type
SavedBundle
}
from
'../src/integrations/meyer/store'
;
import
{
MeyerStore
,
type
SavedBundle
}
from
'../src/integrations/meyer/store'
;
import
{
parseMeyerArgs
,
runMeyerCli
}
from
'../src/cli/meyer-imaging.cli'
;
import
{
parseMeyerArgs
,
runMeyerCli
}
from
'../src/cli/meyer-imaging.cli'
;
import
{
ImageParser
}
from
'../src/modules/sync/pipeline/parsers/image.parser'
;
import
{
ImageParser
}
from
'../src/modules/sync/pipeline/parsers/image.parser'
;
...
@@ -44,7 +44,7 @@ describe('Meyer canonical boundary', () => {
...
@@ -44,7 +44,7 @@ describe('Meyer canonical boundary', () => {
const
facts
=
[...
new
DiagnosisParser
().
parse
(
ctx
),
...
new
ImageParser
().
parse
({
...
ctx
,
canonicalRow
:
bundle
.
image
})];
const
facts
=
[...
new
DiagnosisParser
().
parse
(
ctx
),
...
new
ImageParser
().
parse
({
...
ctx
,
canonicalRow
:
bundle
.
image
})];
for
(
const
fact
of
facts
)
expect
(()
=>
validateFactContent
(
fact
.
type
,
fact
.
subjectId
,
fact
.
content
)).
not
.
toThrow
();
for
(
const
fact
of
facts
)
expect
(()
=>
validateFactContent
(
fact
.
type
,
fact
.
subjectId
,
fact
.
content
)).
not
.
toThrow
();
expect
(
facts
[
0
]
!
.
content
).
toMatchObject
({
code
:
'K02'
,
code_source
:
'image_ai'
,
tooth_position
:
'16'
});
expect
(
facts
[
0
]
!
.
content
).
toMatchObject
({
code
:
'K02'
,
code_source
:
'image_ai'
,
tooth_position
:
'16'
});
expect
(
bundle
.
raw
.
symptom
.
diagnosis
).
toBe
(
'原始综合诊断'
);
expect
(
bundle
.
raw
.
symptom
!
.
diagnosis
).
toBe
(
'原始综合诊断'
);
expect
(
bundle
.
image
.
analysis
).
toEqual
({
source
:
'meyer'
,
summary
:
'原始综合诊断'
,
description
:
'描述'
,
expect
(
bundle
.
image
.
analysis
).
toEqual
({
source
:
'meyer'
,
summary
:
'原始综合诊断'
,
description
:
'描述'
,
analyzedAt
:
null
,
findings
:
bundle
.
diagnoses
.
map
(
d
=>
({
name
:
d
.
name
,
toothPosition
:
d
.
toothPosition
,
code
:
d
.
code
}))
});
analyzedAt
:
null
,
findings
:
bundle
.
diagnoses
.
map
(
d
=>
({
name
:
d
.
name
,
toothPosition
:
d
.
toothPosition
,
code
:
d
.
code
}))
});
expect
(
facts
.
find
(
f
=>
f
.
type
===
FactType
.
IMAGE_RECORD
)
!
.
content
).
toMatchObject
({
analysis
:
bundle
.
image
.
analysis
});
expect
(
facts
.
find
(
f
=>
f
.
type
===
FactType
.
IMAGE_RECORD
)
!
.
content
).
toMatchObject
({
analysis
:
bundle
.
image
.
analysis
});
...
@@ -146,23 +146,22 @@ describe('Meyer collector durable retry and isolation', () => {
...
@@ -146,23 +146,22 @@ describe('Meyer collector durable retry and isolation', () => {
expect
((
await
collectMeyer
(
c
,
api
,
store
,
codes
,
sink
)).
applied
).
toBe
(
1
);
expect
((
await
collectMeyer
(
c
,
api
,
store
,
codes
,
sink
)).
applied
).
toBe
(
1
);
expect
(
sink
.
apply
).
toHaveBeenCalledWith
(
retained
!
.
bundle
);
expect
(
sink
.
apply
).
toHaveBeenCalledWith
(
retained
!
.
bundle
);
expect
((
await
collectMeyer
(
c
,
api
,
store
,
codes
,
sink
)).
alreadyApplied
).
toBe
(
1
);
expect
((
await
collectMeyer
(
c
,
api
,
store
,
codes
,
sink
)).
alreadyApplied
).
toBe
(
1
);
expect
(
sink
.
apply
).
toHaveBeenCalledTimes
(
1
);
expect
(
sink
.
apply
).
toHaveBeenCalledTimes
(
2
);
expect
(
api
.
symptom
).
toHaveBeenCalledTimes
(
1
);
expect
(
api
.
symptom
).
toHaveBeenCalledTimes
(
1
);
const
file
=
join
(
directory
,
`bundle-
${
imageKey
(
c
.
scopes
[
0
]
!
,
'image-1'
)}
.json`
);
const
file
=
join
(
directory
,
`bundle-
${
imageKey
(
c
.
scopes
[
0
]
!
,
'image-1'
)}
.json`
);
expect
((
await
stat
(
file
)).
mode
&
0o777
).
toBe
(
0o600
);
expect
((
await
stat
(
file
)).
mode
&
0o777
).
toBe
(
0o600
);
});
});
test
(
'retrying an older retained bundle does not rewrite its canonical snapshot'
,
async
()
=>
{
test
(
'retrying an older retained bundle does not rewrite its canonical snapshot'
,
async
()
=>
{
const
c
=
meyerConfig
();
const
api
=
source
();
const
c
=
meyerConfig
();
const
api
=
source
();
await
collectMeyer
(
c
,
api
,
store
,
codes
);
const
key
=
`bundle-
${
imageKey
(
c
.
scopes
[
0
]
!
,
'image-1'
)}
`
;
const
key
=
`bundle-
${
imageKey
(
c
.
scopes
[
0
]
!
,
'image-1'
)}
`
;
const
saved
=
(
await
store
.
read
<
SavedBundle
>
(
key
))
!
;
const
saved
:
SavedBundle
=
{
bundle
:
meyerBundle
(),
applied
:
false
}
;
delete
saved
.
bundle
.
image
.
analysis
;
delete
saved
.
bundle
.
image
.
analysis
;
await
store
.
write
(
key
,
saved
);
await
store
.
write
(
key
,
saved
);
const
sink
=
{
apply
:
jest
.
fn
(
async
(
_bundle
:
MeyerBundle
)
=>
{})
};
const
sink
=
{
apply
:
jest
.
fn
(
async
(
_bundle
:
MeyerBundle
)
=>
{})
};
expect
((
await
collectMeyer
(
c
,
api
,
store
,
codes
,
sink
)).
applied
).
toBe
(
1
);
expect
((
await
collectMeyer
(
c
,
api
,
store
,
codes
,
sink
)).
applied
).
toBe
(
1
);
expect
(
sink
.
apply
).
toHaveBeenCalledWith
(
saved
.
bundle
);
expect
(
sink
.
apply
).
toHaveBeenCalledWith
(
saved
.
bundle
);
expect
(
sink
.
apply
.
mock
.
calls
[
0
]
!
[
0
].
image
).
not
.
toHaveProperty
(
'analysis'
);
expect
(
sink
.
apply
.
mock
.
calls
[
0
]
!
[
0
].
image
).
not
.
toHaveProperty
(
'analysis'
);
expect
(
api
.
symptom
).
toHaveBeenCalledTimes
(
1
);
expect
(
api
.
symptom
).
not
.
toHaveBeenCalled
(
);
});
});
test
(
'empty results remain eligible on rerun, never claim a zero-finding completed analysis'
,
async
()
=>
{
test
(
'empty results remain eligible on rerun, never claim a zero-finding completed analysis'
,
async
()
=>
{
const
api
=
source
();
api
.
symptom
=
jest
.
fn
(
async
()
=>
({
teeth
:
[],
diagnosis
:
''
,
description
:
''
}));
const
api
=
source
();
api
.
symptom
=
jest
.
fn
(
async
()
=>
({
teeth
:
[],
diagnosis
:
''
,
description
:
''
}));
...
@@ -170,6 +169,61 @@ describe('Meyer collector durable retry and isolation', () => {
...
@@ -170,6 +169,61 @@ describe('Meyer collector durable retry and isolation', () => {
api
.
symptom
=
jest
.
fn
(
async
()
=>
meyerSymptom
());
api
.
symptom
=
jest
.
fn
(
async
()
=>
meyerSymptom
());
expect
((
await
collectMeyer
(
meyerConfig
(),
api
,
store
,
codes
)).
collected
).
toBe
(
1
);
expect
((
await
collectMeyer
(
meyerConfig
(),
api
,
store
,
codes
)).
collected
).
toBe
(
1
);
});
});
test
(
'image is written before querying unavailable analysis; late analysis links the same image'
,
async
()
=>
{
const
c
=
meyerConfig
(),
api
=
source
();
const
written
:
MeyerBundle
[]
=
[];
const
sink
=
{
apply
:
jest
.
fn
(
async
(
b
:
MeyerBundle
)
=>
{
written
.
push
(
b
);
})
};
api
.
symptom
=
jest
.
fn
(
async
()
=>
{
expect
(
written
).
toHaveLength
(
1
);
expect
(
written
[
0
]
!
.
phase
).
toBe
(
'image'
);
throw
Error
(
'analysis unavailable'
);
});
const
first
=
await
collectMeyer
(
c
,
api
,
store
,
codes
,
sink
);
expect
(
first
).
toMatchObject
({
imagesApplied
:
1
,
failed
:
0
,
analysisFailed
:
1
});
expect
(
written
[
0
]
!
.
image
).
not
.
toHaveProperty
(
'analysis'
);
expect
(
written
[
0
]
!
.
diagnoses
).
toEqual
([]);
const
pending
=
new
MeyerStore
(
join
(
directory
,
'pending'
));
expect
(
await
pending
.
keys
()).
toHaveLength
(
1
);
api
.
symptom
=
jest
.
fn
(
async
()
=>
meyerSymptom
());
const
next
=
await
collectMeyer
(
c
,
api
,
store
,
codes
,
sink
);
expect
(
next
).
toMatchObject
({
imagesApplied
:
0
,
applied
:
1
,
failed
:
0
,
analysisFailed
:
0
});
expect
(
written
).
toHaveLength
(
2
);
expect
(
written
[
1
]
!
.
image
.
externalId
).
toBe
(
written
[
0
]
!
.
image
.
externalId
);
expect
(
written
[
1
]
!
.
diagnoses
.
every
(
d
=>
d
.
sourceImageExternalId
===
written
[
0
]
!
.
image
.
externalId
)).
toBe
(
true
);
expect
(
await
pending
.
keys
()).
toEqual
([]);
expect
((
await
collectMeyer
(
c
,
api
,
store
,
codes
,
sink
)).
alreadyApplied
).
toBe
(
1
);
expect
(
written
).
toHaveLength
(
2
);
});
test
(
'blank analysis keeps a persisted image; missing study time and summary-only analysis do not invent diagnoses'
,
async
()
=>
{
const
c
=
meyerConfig
(),
im
=
{
...
meyerImage
(),
studyTime
:
null
};
const
metadata
=
mapMeyerImage
(
c
,
c
.
scopes
[
0
]
!
,
im
,
'2026-10-17T00:00:00.000Z'
)
!
;
expect
(
metadata
.
image
).
not
.
toHaveProperty
(
'analysis'
);
expect
(
metadata
.
diagnoses
).
toEqual
([]);
const
analysis
=
mapMeyerAnalysis
(
c
,
c
.
scopes
[
0
]
!
,
im
,
meyerSymptom
(),
codes
,
metadata
.
observedAt
)
!
;
expect
(
analysis
.
image
.
analysis
!
.
findings
).
toHaveLength
(
2
);
expect
(
analysis
.
diagnoses
).
toEqual
([]);
expect
(
analysis
.
image
.
analysis
!
.
analyzedAt
).
toBeNull
();
const
summary
=
mapMeyerAnalysis
(
c
,
c
.
scopes
[
0
]
!
,
meyerImage
(),
{
teeth
:
[],
diagnosis
:
'源综合结论'
,
description
:
''
},
codes
,
metadata
.
observedAt
)
!
;
expect
(
summary
.
image
.
analysis
!
.
summary
).
toBe
(
'源综合结论'
);
expect
(
summary
.
diagnoses
).
toEqual
([]);
const
api
=
source
();
api
.
symptom
=
jest
.
fn
(
async
()
=>
({
teeth
:
[],
diagnosis
:
''
,
description
:
''
}));
const
sink
=
{
apply
:
jest
.
fn
(
async
(
_b
:
MeyerBundle
)
=>
{})
};
const
result
=
await
collectMeyer
(
c
,
api
,
store
,
codes
,
sink
);
expect
(
result
).
toMatchObject
({
imagesApplied
:
1
,
deferred
:
1
,
applied
:
0
,
failed
:
0
});
expect
(
sink
.
apply
).
toHaveBeenCalledTimes
(
1
);
const
key
=
imageKey
(
c
.
scopes
[
0
]
!
,
'image-1'
);
expect
(
await
store
.
read
(
`image-
${
key
}
`
)).
toMatchObject
({
applied
:
true
});
expect
(
await
store
.
read
(
`bundle-
${
key
}
`
)).
toBeNull
();
});
test
(
'analysis write failure retries its immutable snapshot without reapplying metadata'
,
async
()
=>
{
const
c
=
meyerConfig
(),
api
=
source
();
const
sink
=
{
apply
:
jest
.
fn
(
async
(
b
:
MeyerBundle
)
=>
{
if
(
b
.
phase
===
'analysis'
)
throw
Error
(
'write failed'
);
})
};
expect
(
await
collectMeyer
(
c
,
api
,
store
,
codes
,
sink
)).
toMatchObject
({
imagesApplied
:
1
,
analysisFailed
:
1
,
failed
:
0
});
api
.
symptom
=
jest
.
fn
(
async
()
=>
{
throw
Error
(
'must use retained analysis'
);
});
const
recovered
=
{
apply
:
jest
.
fn
(
async
(
_b
:
MeyerBundle
)
=>
{})
};
expect
((
await
collectMeyer
(
c
,
api
,
store
,
codes
,
recovered
)).
applied
).
toBe
(
1
);
expect
(
recovered
.
apply
).
toHaveBeenCalledTimes
(
1
);
expect
(
recovered
.
apply
.
mock
.
calls
[
0
]
!
[
0
].
phase
).
toBe
(
'analysis'
);
expect
(
api
.
symptom
).
not
.
toHaveBeenCalled
();
});
test
(
'wrong clinic and changed owner do not produce writes'
,
async
()
=>
{
test
(
'wrong clinic and changed owner do not produce writes'
,
async
()
=>
{
const
api
=
source
();
const
c
=
meyerConfig
();
const
api
=
source
();
const
c
=
meyerConfig
();
await
collectMeyer
(
c
,
api
,
store
,
codes
);
await
collectMeyer
(
c
,
api
,
store
,
codes
);
...
...
apps/pac-service/tests/meyer-resume.spec.ts
View file @
01b9ae78
...
@@ -21,7 +21,7 @@ describe('Meyer conservative limits and recovery', () => {
...
@@ -21,7 +21,7 @@ describe('Meyer conservative limits and recovery', () => {
});
});
test
(
'row checkpoint survives restart; cursor waits for all pages; renewed URLs do not reset progress'
,
async
()
=>
{
test
(
'row checkpoint survives restart; cursor waits for all pages; renewed URLs do not reset progress'
,
async
()
=>
{
const
rows
=
images
(),
c
=
config
();
const
applied
:
string
[]
=
[];
const
rows
=
images
(),
c
=
config
();
const
applied
:
string
[]
=
[];
const
sink
=
{
apply
:
async
(
b
:
MeyerBundle
)
=>
{
applied
.
push
(
b
.
raw
.
image
.
imageId
);
}
};
const
sink
=
{
apply
:
async
(
b
:
MeyerBundle
)
=>
{
if
(
b
.
phase
===
'analysis'
)
applied
.
push
(
b
.
raw
.
image
.
imageId
);
}
};
const
tick
=
()
=>
runMeyerTick
(
c
,
()
=>
source
(
rows
),
new
MeyerStore
(
dir
),
codes
,
sink
,
now
);
const
tick
=
()
=>
runMeyerTick
(
c
,
()
=>
source
(
rows
),
new
MeyerStore
(
dir
),
codes
,
sink
,
now
);
const
first
=
await
tick
();
expect
(
first
.
scan
?.
examined
).
toBe
(
2
);
expect
(
first
.
cursor
).
toBeNull
();
const
first
=
await
tick
();
expect
(
first
.
scan
?.
examined
).
toBe
(
2
);
expect
(
first
.
cursor
).
toBeNull
();
expect
(
await
new
MeyerStore
(
dir
).
read
(
'scan-progress'
)).
toMatchObject
({
page
:
1
,
row
:
2
});
expect
(
await
new
MeyerStore
(
dir
).
read
(
'scan-progress'
)).
toMatchObject
({
page
:
1
,
row
:
2
});
...
@@ -33,7 +33,7 @@ describe('Meyer conservative limits and recovery', () => {
...
@@ -33,7 +33,7 @@ describe('Meyer conservative limits and recovery', () => {
});
});
test
(
'offset shift restarts fixed window and does not skip inserted image'
,
async
()
=>
{
test
(
'offset shift restarts fixed window and does not skip inserted image'
,
async
()
=>
{
const
rows
=
images
(),
c
=
config
(),
written
:
string
[]
=
[];
const
rows
=
images
(),
c
=
config
(),
written
:
string
[]
=
[];
const
sink
=
{
apply
:
async
(
b
:
MeyerBundle
)
=>
{
written
.
push
(
b
.
raw
.
image
.
imageId
);
}
};
const
sink
=
{
apply
:
async
(
b
:
MeyerBundle
)
=>
{
if
(
b
.
phase
===
'analysis'
)
written
.
push
(
b
.
raw
.
image
.
imageId
);
}
};
await
runMeyerTick
(
c
,
()
=>
source
(
rows
),
new
MeyerStore
(
dir
),
codes
,
sink
,
now
);
await
runMeyerTick
(
c
,
()
=>
source
(
rows
),
new
MeyerStore
(
dir
),
codes
,
sink
,
now
);
rows
.
unshift
({
...
meyerImage
(),
imageId
:
'inserted'
});
rows
.
unshift
({
...
meyerImage
(),
imageId
:
'inserted'
});
for
(
let
n
=
0
;
n
<
4
;
n
++
)
{
for
(
let
n
=
0
;
n
<
4
;
n
++
)
{
...
@@ -53,7 +53,7 @@ describe('Meyer conservative limits and recovery', () => {
...
@@ -53,7 +53,7 @@ describe('Meyer conservative limits and recovery', () => {
expect
(
first
.
scan
?.
failed
).
toBe
(
1
);
expect
(
first
.
cursor
).
toBeNull
();
expect
(
first
.
scan
?.
failed
).
toBe
(
1
);
expect
(
first
.
cursor
).
toBeNull
();
down
=
false
;
down
=
false
;
const
next
=
await
runMeyerTick
(
c
,
factory
,
new
MeyerStore
(
dir
),
codes
,
sink
,
now
);
const
next
=
await
runMeyerTick
(
c
,
factory
,
new
MeyerStore
(
dir
),
codes
,
sink
,
now
);
expect
(
next
.
scan
?.
complete
).
toBe
(
true
);
expect
(
sink
.
apply
).
toHaveBeenCalledTimes
(
5
);
expect
(
next
.
scan
?.
complete
).
toBe
(
true
);
expect
(
sink
.
apply
).
toHaveBeenCalledTimes
(
10
);
});
});
test
(
'duplicate caused by offset drift blocks completion and can recover by re-scan'
,
async
()
=>
{
test
(
'duplicate caused by offset drift blocks completion and can recover by re-scan'
,
async
()
=>
{
const
c
=
{
...
config
(),
maxNewImagesPerTick
:
10
},
rows
=
images
();
let
drift
=
true
;
const
c
=
{
...
config
(),
maxNewImagesPerTick
:
10
},
rows
=
images
();
let
drift
=
true
;
...
@@ -66,7 +66,7 @@ describe('Meyer conservative limits and recovery', () => {
...
@@ -66,7 +66,7 @@ describe('Meyer conservative limits and recovery', () => {
expect
(
first
.
scan
?.
failed
).
toBe
(
1
);
expect
(
first
.
cursor
).
toBeNull
();
expect
(
first
.
scan
?.
failed
).
toBe
(
1
);
expect
(
first
.
cursor
).
toBeNull
();
drift
=
false
;
drift
=
false
;
const
next
=
await
runMeyerTick
(
c
,
factory
,
new
MeyerStore
(
dir
),
codes
,
sink
,
now
);
const
next
=
await
runMeyerTick
(
c
,
factory
,
new
MeyerStore
(
dir
),
codes
,
sink
,
now
);
expect
(
next
.
scan
?.
complete
).
toBe
(
true
);
expect
(
sink
.
apply
).
toHaveBeenCalledTimes
(
5
);
expect
(
next
.
scan
?.
complete
).
toBe
(
true
);
expect
(
sink
.
apply
).
toHaveBeenCalledTimes
(
10
);
});
});
test
(
'watch backs off for thrown outages and error exits, recovers, then stops cleanly'
,
async
()
=>
{
test
(
'watch backs off for thrown outages and error exits, recovers, then stops cleanly'
,
async
()
=>
{
const
c
=
{
...
config
(),
pollIntervalMs
:
1000
,
errorBackoffMaxMs
:
4000
};
const
c
=
{
...
config
(),
pollIntervalMs
:
1000
,
errorBackoffMaxMs
:
4000
};
...
...
apps/pac-service/tests/meyer-scheduler.spec.ts
View file @
01b9ae78
...
@@ -23,13 +23,25 @@ describe('independent Meyer scheduling', () => {
...
@@ -23,13 +23,25 @@ describe('independent Meyer scheduling', () => {
symptom
:
async
()
=>
ready
?
meyerSymptom
()
:
{
teeth
:
[],
diagnosis
:
''
,
description
:
''
},
symptom
:
async
()
=>
ready
?
meyerSymptom
()
:
{
teeth
:
[],
diagnosis
:
''
,
description
:
''
},
});
});
const
first
=
await
runMeyerTick
(
c
,
factory
,
store
,
codes
,
sink
,
now
);
const
first
=
await
runMeyerTick
(
c
,
factory
,
store
,
codes
,
sink
,
now
);
expect
(
first
.
pending
).
toBe
(
1
);
expect
(
first
.
scan
?.
failed
).
toBe
(
0
);
expect
(
sink
.
apply
).
not
.
toHaveBeenCalled
();
expect
(
first
.
pending
).
toBe
(
1
);
expect
(
first
.
scan
?.
failed
).
toBe
(
0
);
expect
(
sink
.
apply
).
toHaveBeenCalledTimes
(
1
);
expect
(
first
.
scan
?.
imagesApplied
).
toBe
(
1
);
expect
(
await
store
.
read
(
'cursor'
)).
toEqual
({
through
:
'2026-10-10 10:00:00'
});
expect
(
await
store
.
read
(
'cursor'
)).
toEqual
({
through
:
'2026-10-10 10:00:00'
});
const
key
=
imageKey
(
c
.
scopes
[
0
]
!
,
im
.
imageId
);
const
key
=
imageKey
(
c
.
scopes
[
0
]
!
,
im
.
imageId
);
const
item
=
await
pending
.
read
<
any
>
(
key
);
await
pending
.
write
(
key
,
{
...
item
,
nextRetryAt
:
'2000-01-01T00:00:00Z'
});
const
item
=
await
pending
.
read
<
any
>
(
key
);
await
pending
.
write
(
key
,
{
...
item
,
nextRetryAt
:
'2000-01-01T00:00:00Z'
});
ready
=
true
;
ready
=
true
;
const
next
=
await
runMeyerTick
(
c
,
factory
,
store
,
codes
,
sink
,
now
);
const
next
=
await
runMeyerTick
(
c
,
factory
,
store
,
codes
,
sink
,
now
);
expect
(
next
.
retryAttempted
).
toBe
(
1
);
expect
(
next
.
pending
).
toBe
(
0
);
expect
(
sink
.
apply
).
toHaveBeenCalledTimes
(
1
);
expect
(
next
.
retryAttempted
).
toBe
(
1
);
expect
(
next
.
pending
).
toBe
(
0
);
expect
(
sink
.
apply
).
toHaveBeenCalledTimes
(
2
);
});
test
(
'analysis outage does not block the image scan cursor or later images'
,
async
()
=>
{
const
c
=
config
(),
store
=
new
MeyerStore
(
dir
),
written
:
string
[]
=
[];
const
rows
=
[
meyerImage
(),
{
...
meyerImage
(),
imageId
:
'second'
}];
const
first
=
await
runMeyerTick
(
c
,
()
=>
({
list
:
async
()
=>
({
total
:
2
,
page
:
1
,
size
:
100
,
images
:
rows
}),
symptom
:
async
()
=>
{
throw
Error
(
'analysis service down'
);
},
}),
store
,
codes
,
{
apply
:
async
b
=>
{
written
.
push
(
b
.
image
.
externalId
);
}
},
now
);
expect
(
written
).
toHaveLength
(
2
);
expect
(
first
.
scan
).
toMatchObject
({
complete
:
true
,
imagesApplied
:
2
,
failed
:
0
,
analysisFailed
:
2
});
expect
(
first
.
pending
).
toBe
(
2
);
expect
(
first
.
cursor
).
toEqual
({
through
:
'2026-10-10 10:00:00'
});
});
});
test
(
'failed page never advances cursor; minute overlap is explicit'
,
async
()
=>
{
test
(
'failed page never advances cursor; minute overlap is explicit'
,
async
()
=>
{
const
store
=
new
MeyerStore
(
dir
);
const
windows
:
string
[]
=
[];
const
store
=
new
MeyerStore
(
dir
);
const
windows
:
string
[]
=
[];
...
...
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