Skip to content

P
pac
Commit ed4a22ec by luoqi
Options

feat(meyer): adopt 200-image pages and batch analysis API

parent ff1c0399
Pipeline #3852 failed in 0 seconds
Showing with 233 additions and 51 deletions
apps/pac-service/src/integrations/meyer/README.md
......@@ -51,6 +51,16 @@
从 `apps/pac-service` 执行。凭据只从 `MEYER_APP_KEY`、`MEYER_APP_SECRET` 环境变量读取；必须具备 SaaS 已同步诊所及患者所属应用的读取权限。`baseUrl` 使用 HTTPS，localhost HTTP 仅供本地模拟服务测试。
### 当前接口契约（美亚 v0.4）
- 影像列表使用 `query-by-clinic`，每页默认 200 张，上限 200；每轮处理条数仍受共享扫描额度限制。
- 新扫描和待分析补查均使用 `symptom-batch`，每批最多 200 个 `imageIds`。不再向美亚发送旧的单张病症查询，也不发送患者或诊所参数。PAC 仍按诊所合批，先核对影像身份并保存影像，再查分析。
- 批量结果按 `imageId` 关联，不依赖数组位置。缺项、空分析和单项格式异常分别留底、按原规则补查；重复或额外 ID 拒绝整批，不能错挂到其他患者。
- 已保存的旧分页断点继续使用原 `pageSize` 完成固定窗口，下个窗口才采用新配置。不要删除断点或修改旧页大小。
- 授权按 `accessExpiresAt` 缓存，提前 30 秒在下次调用时更新；优先轮换一次性 refresh token，失败或网络结果不确定时丢弃旧 refresh token，再用应用凭据签发。并发调用及同一客户端的不同时间窗口共享一次换取操作。
- HTTP 401 或封套 `code=401/401001` 仅触发一次重新认证与业务重试。到期时长以美亚返回值为准，不写死为两小时或七天；令牌不写入文件或日志。
- 不依赖 Java SDK，不新增影像事件推送；继续由现有单一定时器轮询。
复制 `config.example.json` 到仓库外，填入允许采集的 clinicIds 和时间范围；namespace 是稳定的应用来源名，不要随运行改变。状态目录按环境和应用隔离，包含患者数据，应受权限及备份策略保护。
```sh
......@@ -69,7 +79,7 @@ pnpm meyer-imaging -- --config=/absolute/meyer.json --state-dir=/absolute/privat
## 本期取舍与上线前核实
- 已用测试患者完成真实只读联调：PDF 三项身份参数可用，分析可落库，文件 ID 可换取 OSS 签名地址。详见 QA 记录。
- 历史联调记录中的单张三项参数属于旧契约；当前按上节 v0.4 批量接口执行。文件 ID 继续按既有接口换取 OSS 签名地址，文件地址接口的上限仍为 100，不能与影像列表/分析的 200 混用。
- 无更新游标、完整快照分页和删除标记，无法承诺完整生命周期同步；分页数量变化、重复/缺页会报错，重跑固定窗口。只在空响应和缺字段时延后，不能把它们当作阴性结论。
- 宿主若以后推送同一批美亚影像，必须先统一来源标识和接入责任，避免与本路线重复。首期应只开明确授权的试点诊所。
- 测试环境曾运行固定窗口联调；这不等于已启用持续任务。启用前应核实实际进程、启动配置、持久状态及扫描水位。开发回归使用本地 HTTP 与独立 PostgreSQL 数据库。
......@@ -166,7 +176,7 @@ pnpm meyer-imaging -- --config=/absolute/meyer.json --state-dir=/absolute/privat
| requestIntervalMs | 500 | 串行请求间隔至少 500ms |
| maxListPagesPerTick | 10 | 新影像扫描每轮最多查询 10 页（含续跑核对）；待处理项逐项也受此上限约束 |
| errorBackoffMaxMs | 3600000 | 连续异常最长等待 1 小时 |
| pageSize | 100 | 每次列表响应最多 100 张；不是每轮处理限额 |
| pageSize | 200 | 每次列表响应最多 200 张；不是每轮处理限额，旧固定窗口断点保留原大小 |
达到分析次数上限后，人工确认需要补查时，可使用同一配置、同一持久状态目录，把固定窗口限定为对应影像入库时刻，再执行默认单次命令（`--apply`，不带 `--tick/--watch`）。这次显式查询可越过等待时间和次数上限；若仍为空或异常，继续保留人工状态。它不绕过影像类型、身份或落库校验，也不重算已完成分析。
......
apps/pac-service/src/integrations/meyer/client.ts
......@@ -3,14 +3,19 @@ import { z } from 'zod';
import type { MeyerConfig, MeyerImage, MeyerScope, MeyerSource } from './contracts';
const PageSchema = z.object({ total: z.number().int().nonnegative(), page: z.number().int().positive(),
size: z.number().int().min(1).max(100), images: z.array(z.unknown()).max(100) });
const TokenSchema = z.object({ accessToken: z.string().min(1), accessExpiresAt: z.string().datetime({ offset: true }) });
size: z.number().int().min(1).max(200), images: z.array(z.unknown()).max(200) });
const TokenSchema = z.object({ accessToken: z.string().min(1), accessExpiresAt: z.string().datetime({ offset: true }),
refreshToken: z.string().min(1).nullish(), refreshExpiresAt: z.string().datetime({ offset: true }).nullish() });
interface TokenSession {
token?: { value: string; expires: number; refresh?: string; refreshExpires?: number };
refreshing?: Promise<string>;
}
/** 只实现三种只读操作；不接入 PAC 主服务，不调用美亚写接口。 */
/** 只读影像、分析与文件查询；同一应用的时间窗口共享授权，不调用美亚写接口。 */
export class MeyerClient implements MeyerSource {
constructor(private readonly config: MeyerConfig,
private readonly credentials: { appKey: string; secret: string },
private readonly session: { token?: { value: string; expires: number } } = {}) {
private readonly session: TokenSession = {}) {
if (!credentials.appKey.trim() || !credentials.secret.trim()) throw new Error('Meyer credentials required');
}
forWindow(config: MeyerConfig): MeyerClient {
......@@ -48,20 +53,45 @@ export class MeyerClient implements MeyerSource {
}
private async getToken(): Promise<string> {
if (this.session.token && this.session.token.expires > Date.now() + 30000) return this.session.token.value;
const result = await this.request('/iam/external/oauth/tokens/pub', {
appKey: this.credentials.appKey, secret: this.credentials.secret, clientType: 'EXTERNAL_APP',
});
if (!this.session.refreshing) {
this.session.refreshing = this.refreshOrIssue().finally(() => { this.session.refreshing = undefined; });
}
return this.session.refreshing;
}
private acceptToken(result: { code: number; data?: unknown }): string {
if (result.code !== 200) throw new Error(`Meyer authentication code=${result.code}`);
const token = TokenSchema.safeParse(result.data);
if (!token.success || Date.parse(token.data.accessExpiresAt) <= Date.now()) throw new Error('Meyer invalid token response');
this.session.token = { value: token.data.accessToken, expires: Date.parse(token.data.accessExpiresAt) };
this.session.token = { value: token.data.accessToken, expires: Date.parse(token.data.accessExpiresAt),
refresh: token.data.refreshToken ?? undefined,
refreshExpires: token.data.refreshExpiresAt ? Date.parse(token.data.refreshExpiresAt) : undefined };
return this.session.token.value;
}
private async refreshOrIssue(): Promise<string> {
const previous = this.session.token;
// 一次性 refreshToken 先移出缓存；网络结果不确定时也不能再次提交旧值。
this.session.token = undefined;
if (previous?.refresh && (previous.refreshExpires === undefined || previous.refreshExpires > Date.now())) {
try {
return this.acceptToken(await this.request('/iam/external/oauth/refreshes/pub', { refreshToken: previous.refresh }));
} catch { /* 不重试旧刷新令牌，回退到应用凭据签发新会话。 */ }
}
const result = await this.request('/iam/external/oauth/tokens/pub', {
appKey: this.credentials.appKey, secret: this.credentials.secret, clientType: 'EXTERNAL_APP',
});
return this.acceptToken(result);
}
private async query(path: string, body: unknown): Promise<unknown> {
let result = await this.request(path, body, await this.getToken());
if (result.code === 401) {
this.session.token = undefined;
result = await this.request(path, body, await this.getToken());
let used = await this.getToken();
let result = await this.request(path, body, used);
if (result.code === 401 || result.code === 401001) {
// 并发返回的旧 401 不能清掉另一请求刚更新的令牌。
if (this.session.token?.value === used) this.session.token.expires = 0;
used = await this.getToken();
result = await this.request(path, body, used);
if ((result.code === 401 || result.code === 401001) && this.session.token?.value === used) {
this.session.token.expires = 0;
}
}
if (result.code !== 200) throw new Error(`Meyer business failure code=${result.code}: ${path}`);
return result.data;
......@@ -87,10 +117,26 @@ export class MeyerClient implements MeyerSource {
}
return result.urls;
}
symptom(image: MeyerImage): Promise<unknown> {
// PDF 比旧 SDK 多 patientId，按文档携带完整归属；上线前需实测服务端版本。
return this.query('/imaging/external/images/symptom', {
imageClinicId: image.clinicId, patientId: image.patientId, imageId: image.imageId,
});
async symptoms(images: readonly MeyerImage[]): Promise<ReadonlyMap<string, unknown>> {
if (images.some(image => !this.config.scopes.some(s => s.clinicId === image.clinicId))) {
throw new Error('analysis clinic outside configured scope');
}
const ids = z.array(z.string().trim().min(1).max(256)).min(1).max(200).parse(images.map(image => image.imageId));
const imageIds = [...new Set(ids)];
const raw = await this.query('/imaging/external/images/symptom-batch', { imageIds });
const rows = z.array(z.object({ imageId: z.string().min(1).max(256) }).passthrough()).max(200).safeParse(raw);
if (!rows.success) throw new Error('Meyer invalid batch symptom response');
const result = new Map<string, unknown>();
for (const row of rows.data) {
if (!imageIds.includes(row.imageId) || result.has(row.imageId)) throw new Error('Meyer ambiguous batch symptom image');
const { imageId, ...symptom } = row;
result.set(imageId, symptom);
}
return result;
}
async symptom(image: MeyerImage): Promise<unknown> {
const result = await this.symptoms([image]);
if (!result.has(image.imageId)) throw new Error('Meyer batch symptom omitted image');
return result.get(image.imageId);
}
}
apps/pac-service/src/integrations/meyer/config.example.json
......@@ -5,7 +5,7 @@
"timezone": "Asia/Shanghai",
"timeStart": "2026-10-01 00:00:00",
"timeEnd": "2026-10-16 23:59:59",
"pageSize": 100,
"pageSize": 200,
"maxPages": 1000,
"timeoutMs": 10000,
"requestIntervalMs": 500,
......
apps/pac-service/src/integrations/meyer/contracts.ts
......@@ -32,7 +32,7 @@ const ConfigFields = z.object({
scopes: z.array(ScopeSchema).min(1).max(100),
timeStart: sourceDate,
timeEnd: sourceDate,
pageSize: z.number().int().min(1).max(100).default(100),
pageSize: z.number().int().min(1).max(200).default(200),
maxPages: z.number().int().min(1).max(10000).default(1000),
timeoutMs: z.number().int().min(100).max(60000).default(10000),
roundTimeoutMs: z.number().int().min(1000).max(86400000).default(600000),
......@@ -98,8 +98,12 @@ export interface MeyerSink { apply(bundle: MeyerBundle): Promise<void> }
export interface MeyerSource {
list(scope: MeyerScope, page: number): Promise<{ total: number; page: number; size: number; images: unknown[] }>;
symptom(image: MeyerImage): Promise<unknown>;
/** 批量按影像 ID 关联；缺项不等于阴性结果。适配旧本地来源时可只实现 symptom。 */
symptoms?(images: readonly MeyerImage[]): Promise<ReadonlyMap<string, unknown>>;
/** 调度预算不足时延后分析，不能虚增实际分析请求次数。 */
canRequestAnalysis?(): boolean;
/** 合批前重读身份时，为已核实影像预留一次分析请求，避免列表耗尽整轮额度。 */
remainingRequests?(): number;
}
export function hash(value: unknown): string {
const stable = (v: unknown): unknown => Array.isArray(v) ? v.map(stable) :
......
apps/pac-service/src/integrations/meyer/scan.ts
......@@ -26,7 +26,7 @@ export async function scanWindow(config: MeyerConfig, sourceFor: (c: MeyerConfig
}
if (progress.version !== 1 || !/^[a-f0-9-]{36}$/.test(progress.generation ?? '') || progress.start !== start || progress.end !== end ||
!Number.isInteger(progress.scopeIndex) || progress.scopeIndex < 0 || progress.scopeIndex > config.scopes.length ||
!Number.isInteger(progress.pageSize) || progress.pageSize < 1 || progress.pageSize > 100 ||
!Number.isInteger(progress.pageSize) || progress.pageSize < 1 || progress.pageSize > 200 ||
!Number.isInteger(progress.row) || progress.row < 0 || !Number.isInteger(progress.page) || progress.page < 1) {
throw new Error('invalid scan checkpoint; do not discard it');
}
......@@ -88,23 +88,30 @@ export async function scanWindow(config: MeyerConfig, sourceFor: (c: MeyerConfig
}
for (const raw of page.images.slice(0, progress.row)) seen.add(ids(raw));
while (progress.row < page.images.length && result.examined < config.maxNewImagesPerTick) {
const image = page.images[progress.row]; result.examined++;
if (seen.has(ids(image))) {
progress.generation = randomUUID(); progress.page = 1; progress.row = 0;
progress.pageData = undefined; progress.total = undefined; progress.firstHash = undefined;
await store.write('scan-progress', progress);
throw new Error('duplicate image across scan pages; restart fixed window');
const count = source.symptoms ? Math.min(page.images.length - progress.row,
config.maxNewImagesPerTick - result.examined) : 1;
const images = page.images.slice(progress.row, progress.row + count);
result.examined += images.length;
for (const image of images) {
if (seen.has(ids(image))) {
progress.generation = randomUUID(); progress.page = 1; progress.row = 0;
progress.pageData = undefined; progress.total = undefined; progress.firstHash = undefined;
await store.write('scan-progress', progress);
throw new Error('duplicate image across scan pages; restart fixed window');
}
seen.add(ids(image));
}
seen.add(ids(image));
const one = await collectMeyer({ ...window, scopes: [scope] }, {
list: async () => ({ total: 1, page: 1, size: window.pageSize, images: [image] }),
list: async () => ({ total: images.length, page: 1, size: window.pageSize, images }),
canRequestAnalysis: () => source.canRequestAnalysis?.() ?? true,
symptom: im => source.symptom(im),
...(source.symptoms ? { symptoms: (items: Parameters<NonNullable<MeyerSource['symptoms']>>[0]) => source.symptoms!(items) } : {}),
}, evidenceStore, codes, sink, true, now);
for (const key of ['seen', 'collected', 'applied', 'alreadyApplied', 'deferred', 'failed', 'imagesApplied', 'analysisFailed'] as const) result[key] += one[key];
result.errors.push(...one.errors);
if (one.failed) return result; // 不跳过未知失败；重试从本行恢复。
progress.row++; await store.write('scan-progress', progress);
// 整组都已落库或留下独立待处理记录后才推进；中断时安全重放这组。
progress.row += images.length; await store.write('scan-progress', progress);
}
if (progress.row === page.images.length) {
await ledger.write(`page-${progress.page}`, page.images.map(ids));
......
apps/pac-service/src/integrations/meyer/scheduler.ts
......@@ -3,7 +3,7 @@ import { rm } from 'node:fs/promises';
import { join } from 'node:path';
import { collectMeyer, type PendingImage } from './collector';
import { MeyerStore, type SavedBundle } from './store';
import { imageKey, sourceTime, type MeyerConfig, type MeyerSource, type MeyerSink, hash, type MeyerScope } from './contracts';
import { imageKey, sourceTime, ImageSchema, type MeyerConfig, type MeyerSource, type MeyerSink, hash, type MeyerScope } from './contracts';
import { analysisCapability } from './retry-policy';
import { clinicStateStore, prepareClinicStates } from './clinic-state';
import { MeyerBudgetExhausted, MeyerTickBudget } from './tick-budget';
......@@ -46,16 +46,30 @@ async function runClinicTick(config: MeyerConfig, sourceFor: (c: MeyerConfig) =>
}
due.sort((a, b) => a.value.nextRetryAt.localeCompare(b.value.nextRetryAt));
let retryFailed = 0, retryAttempted = 0;
const verified: Array<{ key: string; value: PendingImage; config: MeyerConfig; image: unknown }> = [];
const retryFailure = async (key: string, value: PendingImage, error: unknown) => {
retryFailed++;
// 移出队列的人工记录不能复活，落盘后的请求次数也不能回退。
if (await identityReviewStore(store).read(key)) { retiredIdentity++; return; }
const retained = await pending.read<PendingImage>(key) ?? value;
const next = { ...retained, attempts: retained.attempts + 1,
...(error instanceof MeyerIdentityError ? { identityIssue: error.message,
identityAttempts: retained.identityAttempts ?? 1 } : {}),
nextRetryAt: new Date(now + 3600000).toISOString() };
await pending.write(key, next);
if (await retireIdentity(store, key, next, now)) retiredIdentity++;
};
for (const entry of due.slice(0, config.maxRetriesPerTick)) {
const { key } = entry;
let value = entry.value;
if (!available()) break;
retryAttempted++;
if (hash(value.scope) !== hash(config.scopes[0])) throw new Error('pending scope mismatch');
if (!value.image.imageCreateTime) { retryFailed++; continue; }
if (!value.image.imageCreateTime) { retryAttempted++; retryFailed++; continue; }
const t = Date.parse(sourceTime(value.image.imageCreateTime));
const retryConfig = { ...config, scopes: [value.scope], timeStart: chinaTime(t - 1000), timeEnd: chinaTime(t + 1000) };
const client = sourceFor(retryConfig);
if (verified.length && client.symptoms && (client.remainingRequests?.() ?? Infinity) < 2) break;
retryAttempted++;
// 每次重试重新读取影像身份，避免拿旧 patientId 给后来改绑的影像写事实。
try {
if (isIdentityPending(value)) {
......@@ -67,6 +81,9 @@ async function runClinicTick(config: MeyerConfig, sourceFor: (c: MeyerConfig) =>
let total: number | undefined;
for (let page = 1; ; page++) {
if (page > Math.min(config.maxPages, config.maxListPagesPerTick)) throw new Error('pending page limit');
if (verified.length && client.symptoms && (client.remainingRequests?.() ?? Infinity) < 2) {
throw new MeyerBudgetExhausted('Meyer request reserved for verified analysis batch');
}
const batch = await client.list(value.scope, page);
if (batch.page !== page || batch.size !== config.pageSize || (total !== undefined && total !== batch.total) ||
batch.images.length !== Math.min(batch.size, Math.max(0, batch.total - (page - 1) * batch.size))) throw new Error('pending page changed');
......@@ -80,6 +97,15 @@ async function runClinicTick(config: MeyerConfig, sourceFor: (c: MeyerConfig) =>
if (value.image.patientId && (found as { patientId?: string }).patientId !== value.image.patientId) {
throw new MeyerIdentityError('pending image ownership changed');
}
if (client.symptoms) {
// 每张影像都先从源端重读身份，不能为合批放宽原来的归属和时间校验。
const image = ImageSchema.parse(found);
if (image.clinicId !== value.scope.clinicId) throw new Error('pending image clinic mismatch');
if (image.imageCreateTime && (sourceTime(image.imageCreateTime) < sourceTime(retryConfig.timeStart) ||
sourceTime(image.imageCreateTime) > sourceTime(retryConfig.timeEnd))) throw new Error('pending image outside requested window');
verified.push({ key, value, config: retryConfig, image });
continue;
}
const result = await collectMeyer(retryConfig, {
list: async () => ({ total: 1, page: 1, size: config.pageSize, images: [found] }),
canRequestAnalysis: () => client.canRequestAnalysis?.() ?? true,
......@@ -89,17 +115,28 @@ async function runClinicTick(config: MeyerConfig, sourceFor: (c: MeyerConfig) =>
if (await identityReviewStore(store).read(key)) retiredIdentity++;
} catch (error) {
if (error instanceof MeyerBudgetExhausted) break;
retryFailed++;
// 条目可能已移出队列，随后写摘要或 sync 失败；不能用旧值把人工记录重新激活。
if (await identityReviewStore(store).read(key)) { retiredIdentity++; continue; }
// collector 可能已持久化请求次数后才在写运行摘要时失败；不能回写旧次数。
const retained = await pending.read<PendingImage>(key) ?? value;
const next = { ...retained, attempts: retained.attempts + 1,
...(error instanceof MeyerIdentityError ? { identityIssue: error.message,
identityAttempts: retained.identityAttempts ?? 1 } : {}),
nextRetryAt: new Date(now + 3600000).toISOString() };
await pending.write(key, next);
if (await retireIdentity(store, key, next, now)) retiredIdentity++;
await retryFailure(key, value, error);
}
}
// 只合并同一诊所、本轮已重新核实的条目；整轮重试名额与业务请求预算仍独立生效。
for (let offset = 0; offset < verified.length; offset += config.pageSize) {
const items = verified.slice(offset, offset + config.pageSize);
const retryConfig = { ...config,
timeStart: items.map(item => item.config.timeStart).sort()[0]!,
timeEnd: items.map(item => item.config.timeEnd).sort().at(-1)! };
const client = sourceFor(retryConfig);
try {
const result = await collectMeyer(retryConfig, {
list: async () => ({ total: items.length, page: 1, size: config.pageSize, images: items.map(item => item.image) }),
canRequestAnalysis: () => client.canRequestAnalysis?.() ?? true,
symptom: image => client.symptom(image),
symptoms: images => client.symptoms!(images),
}, store, codes, sink, true, now, true);
retryFailed += result.failed + result.analysisFailed;
for (const item of items) if (await identityReviewStore(store).read(item.key)) retiredIdentity++;
} catch (error) {
if (error instanceof MeyerBudgetExhausted) break;
for (const item of items) await retryFailure(item.key, item.value, error);
}
}
let scan: Awaited<ReturnType<typeof scanWindow>> | undefined;
......
apps/pac-service/src/integrations/meyer/tick-budget.ts
......@@ -11,6 +11,7 @@ export class MeyerTickBudget {
source(factory: (config: MeyerConfig) => MeyerSource, allowance: number) {
let used = 0;
const available = () => used < allowance && this.available();
const remaining = () => Date.now() >= this.deadline ? 0 : Math.max(0, Math.min(allowance - used, this.limit - this.requests));
const spend = () => {
if (!available()) throw new MeyerBudgetExhausted('Meyer tick budget exhausted');
used++; this.requests++;
......@@ -20,7 +21,11 @@ export class MeyerTickBudget {
return {
list: async (scope, page) => { spend(); return source.list(scope, page); },
canRequestAnalysis: available,
remainingRequests: remaining,
symptom: async image => { spend(); return source.symptom(image); },
...(source.symptoms ? { symptoms: async (images: Parameters<NonNullable<MeyerSource['symptoms']>>[0]) => {
spend(); return source.symptoms!(images);
} } : {}),
};
} };
}
......
apps/pac-service/tests/meyer-imaging-fixtures.ts
......@@ -5,7 +5,7 @@ import { loadDiagnosisCodes, mapMeyer } from '../src/integrations/meyer/mapper';
export const meyerConfig = (): MeyerConfig => ConfigSchema.parse({ baseUrl: 'https://example.com/api/v1',
namespace: 'test', host: 'friday', timezone: 'Asia/Shanghai', scopes: [
{ clinicId: 'clinic-a', tenantId: 'friday-market', sourceUnit: 'brand-a' }],
timeStart: '2026-10-01 00:00:00', timeEnd: '2026-10-16 23:59:59', requestIntervalMs: 0 });
timeStart: '2026-10-01 00:00:00', timeEnd: '2026-10-16 23:59:59', pageSize: 100, requestIntervalMs: 0 });
export const meyerImage = () => ImageSchema.parse({ imageId: 'image-1', clinicId: 'clinic-a', patientId: 'patient-1',
imageCreateTime: '2026-10-10 10:00:00', studyTime: '2026-10-09 09:00:00', modality: 'pano',
imageUrl: 'https://example.com/temporary?signature=abc', imageOssId: 'oss-1' });
......
apps/pac-service/tests/meyer-imaging-postgres.spec.ts
......@@ -12,6 +12,7 @@ import { MeyerCliModule } from '../src/integrations/meyer/meyer-cli.module';
import { MeyerPacSink } from '../src/integrations/meyer/pac-sink';
import { ParserPipeline } from '../src/modules/sync/pipeline/parser-pipeline.service';
import { collectMeyer } from '../src/integrations/meyer/collector';
import { MeyerClient } from '../src/integrations/meyer/client';
import { MeyerStore } from '../src/integrations/meyer/store';
import { mapMeyerImage, mapMeyerAnalysis } from '../src/integrations/meyer/mapper';
import { imageKey } from '../src/integrations/meyer/contracts';
......@@ -70,6 +71,38 @@ describePg('Meyer | isolated PostgreSQL + actual minimal CLI module', () => {
clinicId: 'clinic-a', action: 'visit_registered', subjectType: 'visit', subjectId: 'host-visit',
rawPayload: {}, payloadHash: 'test', occurredAt: new Date() } });
}
test('batch HTTP → canonical facts keeps images when individual results are empty or omitted, and replays safely', async () => {
const directory = await mkdtemp(join(tmpdir(), 'meyer-batch-pg-'));
const state = new MeyerStore(directory), images = [0, 1, 2].map(n => ({ ...meyerImage(), imageId: `batch-${n}` }));
let batches = 0;
const server = createServer(async (req, res) => {
for await (const _chunk of req) { /* 不记录凭据或患者原文。 */ }
const data = req.url?.endsWith('/tokens/pub') ? { accessToken: 'local', accessExpiresAt: new Date(Date.now() + 3600000).toISOString() }
: req.url?.endsWith('/query-by-clinic') ? { total: 3, page: 1, size: 200, images }
: (batches++, [{ imageId: 'batch-1', teeth: [], diagnosis: '', description: '' },
{ imageId: 'batch-0', ...meyerSymptom() }]);
res.setHeader('Content-Type', 'application/json'); res.end(JSON.stringify({ code: 200, data }));
});
try {
await new Promise<void>(r => server.listen(0, '127.0.0.1', r));
const c = { ...meyerConfig(), pageSize: 200, baseUrl: `http://127.0.0.1:${(server.address() as { port: number }).port}/api/v1` };
const client = new MeyerClient(c, { appKey: 'local', secret: 'local' });
const first = await collectMeyer(c, client, state, codes, sink, true);
expect(first).toMatchObject({ imagesApplied: 3, applied: 1, analysisFailed: 1, failed: 0 });
const facts = await prisma.patientFact.findMany({ where: { hostId, patientId, status: 'active' }, orderBy: { subjectId: 'asc' } });
expect(facts.filter(f => f.type === FactType.IMAGE_RECORD)).toHaveLength(3);
expect(facts.filter(f => f.type === FactType.DIAGNOSIS_RECORD)).toHaveLength(2);
const pending = new MeyerStore(join(directory, 'pending'));
expect(await pending.keys()).toHaveLength(2);
await collectMeyer(c, client, state, codes, sink, true);
expect(await prisma.patientFact.findMany({ where: { hostId, patientId, status: 'active' }, orderBy: { subjectId: 'asc' } })).toEqual(facts);
expect(batches).toBe(1);
expect(await prisma.followupPlan.count({ where: { hostId } })).toBe(0);
} finally {
server.closeAllConnections(); await new Promise<void>(r => server.close(() => r()));
await rm(directory, { recursive: true, force: true });
}
});
test('identity recovery CLI verifies live binding and full PAC identity, then resumes the normal tick', async () => {
await hostRecord();
const directory = await mkdtemp(join(tmpdir(), 'meyer-identity-pg-'));
......@@ -79,7 +112,7 @@ describePg('Meyer | isolated PostgreSQL + actual minimal CLI module', () => {
for await (const _chunk of req) { /* 不记录凭据或请求原文。 */ }
res.setHeader('content-type', 'application/json');
const data = req.url?.includes('/tokens/pub') ? { accessToken: 'local-token', accessExpiresAt: new Date(Date.now() + 3600000).toISOString() }
: req.url?.endsWith('/query-by-clinic') ? { total: 1, page: 1, size: 100, images: [image] } : meyerSymptom();
: req.url?.endsWith('/query-by-clinic') ? { total: 1, page: 1, size: 100, images: [image] } : [{ imageId: 'image-1', ...meyerSymptom() }];
res.end(JSON.stringify({ code: 200, data }));
});
const previousKey = process.env.MEYER_APP_KEY, previousSecret = process.env.MEYER_APP_SECRET;
......@@ -350,7 +383,7 @@ describePg('Meyer | isolated PostgreSQL + actual minimal CLI module', () => {
accessToken: 'test-token', accessExpiresAt: new Date(Date.now() + 600000).toISOString(),
} : req.url!.endsWith('/query-by-clinic') ? {
total: 1, page: 1, size: 100, images: [meyerImage()],
} : meyerSymptom();
} : [{ imageId: 'image-1', ...meyerSymptom() }];
res.writeHead(200, { 'Content-Type': 'application/json' }); res.end(JSON.stringify({ code: 200, data }));
});
const previousKey = process.env.MEYER_APP_KEY; const previousSecret = process.env.MEYER_APP_SECRET;
......
apps/pac-service/tests/meyer-imaging.spec.ts
......@@ -273,7 +273,7 @@ describe('Meyer actual HTTP contract', () => {
});
afterEach(async () => { server.closeAllConnections(); await new Promise<void>(r => server.close(() => r())); });
const auth = () => ({ code: 200, data: { accessToken: 'test-token', accessExpiresAt: new Date(Date.now() + 600000).toISOString() } });
test('authentication, complete patient ownership request, and HTTP-200 business errors', async () => {
test('authentication, image ID batch request, and HTTP-200 business errors', async () => {
const calls: string[] = [];
handler = (url, body, token) => {
calls.push(url);
......@@ -286,7 +286,8 @@ describe('Meyer actual HTTP contract', () => {
expect(body).toMatchObject({ clinicId: 'clinic-a', page: 1, size: 100, timeStart: config.timeStart });
return { body: { code: 200, data: { total: 1, page: 1, size: 100, images: [meyerImage()] } } };
}
expect(body).toEqual({ imageClinicId: 'clinic-a', patientId: 'patient-1', imageId: 'image-1' });
expect(url.endsWith('/symptom-batch')).toBe(true);
expect(body).toEqual({ imageIds: ['image-1'] });
return { body: { code: 500, msg: 'private patient info' } };
};
const client = new MeyerClient(config, { appKey: 'key', secret: 'secret' });
......@@ -333,7 +334,7 @@ describe('Meyer actual HTTP contract', () => {
const stdout = jest.spyOn(console, 'log').mockImplementation(() => {});
try {
handler = url => ({ body: url.endsWith('/tokens/pub') ? auth() : { code: 200,
data: url.endsWith('/query-by-clinic') ? { total: 1, page: 1, size: 100, images: [meyerImage()] } : meyerSymptom() } });
data: url.endsWith('/query-by-clinic') ? { total: 1, page: 1, size: 100, images: [meyerImage()] } : [{ imageId: 'image-1', ...meyerSymptom() }] } });
const { writeFile } = await import('node:fs/promises');
await writeFile(join(dir, 'config.json'), JSON.stringify((({ scopes, ...rest }) => ({ ...rest, clinicIds: scopes.map(s => s.clinicId) }))(config)));
expect(await runMeyerCli([`--config=${join(dir, 'config.json')}`, `--state-dir=${join(dir, 'state')}`])).toBe(0);
......
docs/qa/2026-10-21-meyer-batch-contract.md 0 → 100644
# 美亚 v0.4：200 条分页、批量分析与授权更新
## 范围
- 分支：`codex/meyer-clinic-isolation`，基线 `ff1c039`；本轮修改未提交、未合并、未部署 PAC。
- 依据：2026-10-21 提供的《PACS_瑞尔接口对接文档.md》v0.4 与 Java SDK；文件指纹见 `output/meyer-batch-20261021/vendor-inputs.json`。
- PAC 保持直接 HTTP 对接，不引入 JAR。按本轮确认不接入影像事件推送，沿用单一定时器。
## 改动
- `query-by-clinic` 的配置默认值及响应校验上限改为 200；现存 100 条分页断点完成后才切换新大小，不删除游标或留底。
- 新扫描和到期补查使用 `symptom-batch`，只传 `imageIds`，每批最多 200。同一诊所先完成影像身份校验和影像保存，再合并分析请求。
- 结果按影像 ID 对应。省略项、空分析、单项格式错误分别留待补查；额外或重复 ID 拒绝整批。分析次数与补查时间仍逐张持久化，已应用快照不重复拉取。
- 保留原诊所隔离、整轮额度、分析降频/上限和身份异常最多 8 次补查。身份重读阶段为已核实影像预留一次分析请求，避免小额度只反复查列表。
- access token 按返回的到期时间提前 30 秒更新。共享同一客户端的并发换取；优先刷新，丢弃结果不确定的一次性 refresh token 后回退应用凭据签发。HTTP 401、业务码 401/401001 最多触发一次重认证与重试。
- 原标准字段、临床映射、时区、FactWriter 和画像/召回判断未调整。文件 URL 接口上限仍为 100，不随影像列表和分析扩到 200。
## 美亚测试接口实测
2026-10-21 14:54（北京），使用当前测试配置访问 `https://re-pacs.myyun.com/api/v1`，仅查询及签发/刷新测试会话，没有写入 PAC 或修改美亚业务数据。
| 项目 | 结果 |
|---|---|
| 列表请求每页 200 | 返回 `size=200`；本次诊所和时间范围共 58 张 |
| 批量 58 张已有影像及 1 个不存在 ID | 返回 58 项，不存在 ID 被省略 |
| 非空分析 | 本次返回 2 项；不据空视图认定阴性或分析已完成 |
| 抽取 3 张全景片与单张接口比较 | 结果一致 |
| 临近过期刷新 | refresh token 正常轮换；新 access token 剩余约 7200 秒 |
脱敏结果：`output/meyer-batch-20261021/vendor-validation.json`。实测只代表当前测试配置，不推断生产是否共用同一美亚应用。
## 本地验证
- 新增批量/授权用例 14 项，覆盖 200 张合批、缺项、重复/额外 ID、空和异常结果、并发授权、401001、刷新网络异常、旧分页断点、写入恢复及极小请求额度。
- 原有影像摄入、诊所隔离、身份恢复、补查上限和文件访问回归通过；诊断映射 40 项通过。
- 独立本地 PostgreSQL 16 项通过：批量 HTTP → 归一 → 真实事实写入；影像先保存、晚到分析更新同一影像、空/缺项保留补查、重复执行事实不变。临时测试库已清理。
- 后端源码及测试类型检查、构建、差异格式检查通过。
测试日志与构建日志在 `output/meyer-batch-20261021/`。未修改测试/生产 PAC 的采集配置、定时器、业务数据或部署版本。
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment