Skip to content

P
pac
Commit b3e3a6c2 by luoqi
Options

feat(friday): 结算WHERE迁移单一真理源 + 冷导入网页通道 + 数据对账 

- 结算/病历 WHERE 从导出侧迁入 manifest transforms(单一真理源):宿主推原表全 status,
  PAC 侧 filter 切消费/退费/退费明细;transforms 新增数值算子 lt/lte/gt/gte
- 退费明细身份纠正(S.3.1):spec.patient_id 部分品牌是诊所本地 id 不可信,
  改从结算头 settlement_id→patient_id 继承(元和王永实测 26421 行全不符)
- 冷导入网页上传通道:上传 zip → yauzl 白名单解压 → 迷你 data 根(含 _shared)→
  队列 cold-import;宽容缺独立表 + 成组完整性校验 + dry-run 预检
- 数据对账:日报复用摄入查询做「源 ↔ PAC 去重患者数」比对(clickhouse countDistinct)
- 测试 18 绿:friday-settlement-where / filter-numeric / cold-import-extract / cold-import-groups

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
parent 2e31eeb7
Showing with 1321 additions and 22 deletions
apps/pac-service/data/friday/assemblers/emr.yaml
......@@ -10,7 +10,7 @@ emits:
occurredAtField: submittedAt
primary:
table: med_emr_info
table: _emr_official # transforms 产出:med_emr_info status∈{3,4}(排草稿;迁自导出侧过滤)
key: emr_sub_id # 小病历号 uuid(两侧连接键 + 临床事实锚点)
dedup_by: emr_sub_id
......
apps/pac-service/data/friday/assemblers/refund_full.yaml
......@@ -13,7 +13,7 @@ emits:
kind: actual
primary:
table: patient_settlement_refund
table: refund_full_rows # transforms 产出:status=4 ∪ (status=3 且 receivable<0)
key: uuid
dedup_by: uuid
......
apps/pac-service/data/friday/assemblers/refund_item.yaml
......@@ -11,7 +11,7 @@ emits:
kind: actual
primary:
table: patient_settlement_spec_refund
table: refund_item_rows # transforms 产出:patient_settlement_spec 里 is_refund=1
key: id
dedup_by: id
......@@ -19,7 +19,8 @@ field_mapping:
externalId: id
updatedAt: updated_gmt_at
createdAt: created_gmt_at
patientExternalId: patient_id
# spec.patient_id 部分品牌是诊所本地 id(不可信)→ manifest S.3.1 已从父结算头 lookup 纠正
patientExternalId: owner_patient_id
clinicId: organization_id
refundedAt: created_gmt_at
amount: net_receipts_this
......
apps/pac-service/data/friday/export.sh
......@@ -142,8 +142,7 @@ mysql_csv "SELECT id,tenant_id,customer_id,treat_plan_id,tooth_position,mode_nam
# ⚠️ 2ac96f6d 品牌不用 status=4:整单冲减直接记 status=3 负金额(实测 200 行)→ 按金额正负切分:
# 消费 = status∈{1,3} 且金额≥0;整单退费 = status=4 或 status=3 负金额(退费第三表达)。
# 其余 status(0 草稿/2/5/6/7/8)语义未明,先不收 → WHERE 收窄(host 等价 dump 允许的过滤)
mysql_csv "SELECT uuid,tenant_id,organization_id,patient_id,doctor_id,status,receivable_this,net_receipts_this,billing_date,registration_id,ref_settlement_id,settlement_serial_num,reason,created_gmt_at,updated_gmt_at FROM \`arrail-settlement-server\`.patient_settlement WHERE status IN (1,3) AND receivable_this >= 0 $(pfa patient_id)" patient_settlement.csv
mysql_csv "SELECT uuid,tenant_id,organization_id,patient_id,doctor_id,status,receivable_this,net_receipts_this,billing_date,registration_id,ref_settlement_id,settlement_serial_num,reason,created_gmt_at,updated_gmt_at FROM \`arrail-settlement-server\`.patient_settlement WHERE (status=4 OR (status=3 AND receivable_this < 0)) $(pfa patient_id)" patient_settlement_refund.csv
mysql_csv "SELECT uuid,tenant_id,organization_id,patient_id,doctor_id,status,receivable_this,net_receipts_this,billing_date,registration_id,ref_settlement_id,settlement_serial_num,reason,created_gmt_at,updated_gmt_at FROM \`arrail-settlement-server\`.patient_settlement $(pf patient_id)" patient_settlement.csv
# 退费行级明细(部分退费轨道:is_refund=1 全部归属 status=3 头单)
mysql_csv "SELECT id,tenant_id,organization_id,patient_id,settlement_id,cure_name,service_project_name,receivable_this,net_receipts_this,is_refund,created_gmt_at,updated_gmt_at FROM \`arrail-settlement-server\`.patient_settlement_spec WHERE is_refund=1 $(pfa patient_id)" patient_settlement_spec_refund.csv
# 支付通道长表(每单×通道一行;lookup 挑金额最大者为主导通道)
......@@ -174,12 +173,12 @@ docker run --rm ${MONGO_ARGS[@]+"${MONGO_ARGS[@]}"} mongo:7 mongosh "$FRIDAY_MON
// (CSV 源天然全字符串;JSON 导出须显式对齐,否则 patientExternalId 等校验失败)
const sid = (v) => (v === undefined || v === null || v === "") ? null : String(flat(v));
const q = { status: { $in: [3, 4] } };
if (process.env.COHORT === "1") {
try {
// cohort 激活但文件读不到 → 硬失败(宁可让 .tmp 校验拒绝,不能静默退化成全量导出)
const ids = JSON.parse(require("fs").readFileSync("/cohort_ids.json", "utf-8"));
const both = (s) => { const n = Number(s); return Number.isFinite(n) && String(n) === s ? [s, n] : [s]; };
q.patientId = { $in: ids.flatMap(both) };
}
} catch (e) {}
let first = true;
print("[");
db.med_emr_info.find(q).forEach((d) => {
......
apps/pac-service/data/friday/manifest.yaml
......@@ -66,11 +66,13 @@ tables:
- { table: med_check, file: med_check.csv } # 影像档案(挂 emr_id 病历号)
- { table: std_check_class, file: std_check_class.csv } # 影像类型字典(class_code→class_name)
- { table: std_diag, file: std_diag.csv } # 诊断字典(linkCode→std_code;K 码用)
- { table: customer_treat_plan, file: customer_treat_plan.csv } # 治疗计划头(方案名/期望)
- { table: customer_treat_plan_item, file: customer_treat_plan_item.csv } # 治疗计划行(牙位/术式/价格)
- { table: patient_settlement, file: patient_settlement.csv } # 消费头单(status 1/3 且金额≥0)
- { table: patient_settlement_refund, file: patient_settlement_refund.csv } # 整单退费(status=4 或 3 负额)
- { table: patient_settlement_spec_refund, file: patient_settlement_spec_refund.csv } # 行级退费明细(is_refund=1)
# 治疗计划 = 头+行强成组(item 是 primary,lookup 头表拿诊所/方案名;缺任一 → planned 产 0 行)。
# group 只约束网页上传(要么全在要么整组跳);CLI 全量不受此限。
- { table: customer_treat_plan, file: customer_treat_plan.csv, group: treatment_plan } # 治疗计划头(方案名/期望)
- { table: customer_treat_plan_item, file: customer_treat_plan_item.csv, group: treatment_plan } # 治疗计划行(牙位/术式/价格)
# 结算原表(全 status,导出侧不做业务 WHERE)—— 消费/退费/明细全部由 PAC transforms 切分(单一真理源)
- { table: patient_settlement, file: patient_settlement.csv } # 结算头单原表(全 status)
- { table: patient_settlement_spec, file: patient_settlement_spec.csv } # 结算明细原表(全量,含 is_refund)
- { table: settlement_modes, file: settlement_modes.csv } # 支付通道长表
- { table: customer_referee_circle, file: customer_referee_circle.csv } # 转介绍圈(患者-患者边)
- { table: customer_consult, file: customer_consult.csv } # 咨询(意向/未成交原因)
......@@ -126,9 +128,16 @@ transforms:
# ── E.1 病历自由文本行(emr.yaml 直接消费 med_emr_info,无需 project:
# assembler field_mapping 天然只取所需列,数组列不映射即不入 canonical)──
# ── E.0 病历排草稿:status∈{3,4} 正式病历(迁自 mongo 导出侧过滤;PAC 侧统一)──
- kind: filter
input: med_emr_info
output: _emr_official
where:
status: { in: ['3', '4'] }
# ── E.2 diag[] 拆行 → 诊断候选 ──
- kind: split_json_array
input: med_emr_info
input: _emr_official
output: _diag_raw
array_field: diag
parent_keys:
......@@ -190,7 +199,7 @@ transforms:
# 注:treat 填充仅 25%(dispose 98% 是处置叙述,整段进 emr_record.disposal 不丢);
# 若后续漏召,可加 dispose keyword 路由补 actual(Phase 2 评估)。
- kind: split_json_array
input: med_emr_info
input: _emr_official
output: _treat_raw
array_field: treat
parent_keys:
......@@ -342,6 +351,7 @@ transforms:
output: _pay_hdr
where:
status: { in: ['1', '3'] }
receivable_this: { gte: 0 } # 应收≥0(迁自导出侧 WHERE;数值算子)
- kind: lookup
input: _pay_hdr
output: payment_rows
......@@ -353,8 +363,43 @@ transforms:
select:
method: modes_name # 主导支付通道(现金/微信/刷卡/医保…干净中文名,原样入 fact)
# ── S.2 退费①:整单冲减(status=4 或 status=3 负金额;导出侧 WHERE 已切分,assembler 直读
# patient_settlement_refund 表;金额负值由 refund parser Math.abs 归一)──
# ── S.2 退费①:整单冲减 —— PAC 侧切分(不再靠导出 WHERE)。两种表达 union:
# ① status=4 整单反向;② status=3 且 receivable_this<0(2ac96f6d 品牌用负额表达退费)。
# 金额负值由 refund parser Math.abs 归一为正 cents。
- kind: filter
input: patient_settlement
output: _refund_status4
where:
status: { equals: '4' }
- kind: filter
input: patient_settlement
output: _refund_neg3
where:
status: { equals: '3' }
receivable_this: { lt: 0 } # 数值算子:负额退费表达
- kind: union
inputs: ['_refund_status4', '_refund_neg3']
output: refund_full_rows
# ── S.3 退费明细②:行级退费(patient_settlement_spec is_refund=1)——PAC 侧 filter ──
- kind: filter
input: patient_settlement_spec
output: _refund_item_raw
where:
is_refund: { equals: '1' }
# S.3.1 身份纠正:spec.patient_id 部分品牌是「诊所本地 id」而非全局 customer id
# (元和王永 691828a5 实测:全 26421 行 spec.patient_id ≠ 结算头 patient_id,如 spec=2/头=666716;
# 全库 2.64 万不符行几乎全出自该诊所)。退费明细本就是结算头的子行 → 患者身份从父头单继承:
# settlement_id → patient_settlement.patient_id(权威,与 payment/refund_full 同源)。
# 命中即覆盖(lookup 语义);未命中保留原值 → 失败行可见,不会静默挂错人。
- kind: lookup
input: _refund_item_raw
output: refund_item_rows
from: patient_settlement
left_key: settlement_id
right_key: uuid
select:
owner_patient_id: patient_id
# ═══════════ 患者关系链(customer_referee_circle → patient_relations 独立表)═══════════
# 宿主 referee_relationship 是字典 id(DB 无字典表,疑似代码硬编码)。解码 = 统计推断
......
apps/pac-service/package.json
......@@ -90,6 +90,7 @@
"socket.io": "^4.8.3",
"winston": "^3.19.0",
"ws": "^8.21.0",
"yauzl": "^3.3.0",
"zod": "^4.4.3"
},
"devDependencies": {
......
apps/pac-service/src/modules/admin/admin.module.ts
......@@ -7,6 +7,7 @@ import { MappingMissAdminController } from './mapping-miss.controller';
import { HostsService } from './hosts.service';
import { AiInvocationsService } from './ai-invocations.service';
import { MappingMissService } from './mapping-miss.service';
import { ImportUploadService } from './import-upload.service';
@Module({
controllers: [
......@@ -14,7 +15,7 @@ import { MappingMissService } from './mapping-miss.service';
AiInvocationsAdminController,
MappingMissAdminController,
],
providers: [HostsService, AiInvocationsService, MappingMissService],
providers: [HostsService, AiInvocationsService, MappingMissService, ImportUploadService],
exports: [HostsService, AiInvocationsService],
})
export class AdminModule {}
apps/pac-service/src/modules/admin/dto/admin.dto.ts
import { createZodDto } from 'nestjs-zod';
import { z } from 'zod';
import {
ListHostsResponseSchema,
HostSummarySchema,
......@@ -32,3 +33,17 @@ export class AiInvocationListQueryDto extends createZodDto(AiInvocationListQuery
export class AiInvocationDetailDto extends createZodDto(AiInvocationDetailSchema) {}
export class AiInvocationStatsDto extends createZodDto(AiInvocationStatsSchema) {}
export class ListAiInvocationsResponseDto extends createZodDto(ListAiInvocationsResponseSchema) {}
// Cold-import 网页上传(host-self)
const ColdImportUploadResponseSchema = z.object({ jobId: z.string() });
export class ColdImportUploadResponseDto extends createZodDto(ColdImportUploadResponseSchema) {}
// 进度:BullMQ job 状态 + 完成时的 ImportRunResult(含 totals);job 已清理/不存在 → state='not_found'
const ColdImportStatusSchema = z.object({
state: z.string(),
result: z.any().nullable(),
failedReason: z.string().nullable(),
// 运行中的真实进度(cohortDone/cohortTotal 批次);非运行态或 dryRun 为 null
progress: z.object({ done: z.number(), total: z.number() }).nullable(),
});
export class ColdImportStatusDto extends createZodDto(ColdImportStatusSchema) {}
apps/pac-service/src/modules/admin/host-self.controller.ts
......@@ -6,16 +6,23 @@ import {
Param,
Patch,
Post,
UploadedFile,
UseInterceptors,
} from '@nestjs/common';
import { ApiBearerAuth, ApiOperation, ApiTags } from '@nestjs/swagger';
import { FileInterceptor } from '@nestjs/platform-express';
import { ApiBearerAuth, ApiConsumes, ApiOperation, ApiTags } from '@nestjs/swagger';
import { ZodResponse } from 'nestjs-zod';
import { Permission } from '@pac/types';
import { ApiCode, Permission } from '@pac/types';
import { RequirePermission } from '../../common/decorators/permissions.decorator';
import { CurrentUser, AuthenticatedUser } from '../../common/decorators/current-user.decorator';
import { BizError } from '../../common/errors/biz-error';
import {
TenantScope,
TenantScopeContext,
} from '../../common/decorators/tenant-scope.decorator';
import {
ColdImportStatusDto,
ColdImportUploadResponseDto,
HostDetailDto,
HostStatsDto,
HostSummaryDto,
......@@ -26,6 +33,16 @@ import {
UpdateHostRequestDto,
} from './dto/admin.dto';
import { HostsService } from './hosts.service';
import { ImportUploadService } from './import-upload.service';
/// multer 内存模式的最小文件形状(不引 @types/multer,同 assistant.controller)
interface UploadedZip {
buffer: Buffer;
originalname: string;
mimetype: string;
}
/// 上传大小上限:存量分批包压缩后通常几十 MB;留 500MB 余量
const MAX_IMPORT_ZIP_BYTES = 500 * 1024 * 1024;
/**
* Host Self Admin — 宿主自己的 admin 用户管理"自己这个 host"的配置页。
......@@ -42,7 +59,10 @@ import { HostsService } from './hosts.service';
@Controller('admin/host/self')
@RequirePermission(Permission.PLATFORM_MANAGE)
export class HostSelfAdminController {
constructor(private readonly hosts: HostsService) {}
constructor(
private readonly hosts: HostsService,
private readonly importUpload: ImportUploadService,
) {}
@Get()
@ZodResponse({ status: 200, type: HostDetailDto })
......@@ -125,4 +145,38 @@ export class HostSelfAdminController {
) {
return this.hosts.removeCallbackSecretBySuffix(scope.hostId, suffix);
}
// ─── 数据导入(上传压缩包 → 后台 cold-import,方向:宿主 → PAC)───
@Post('cold-import')
@ApiConsumes('multipart/form-data')
@UseInterceptors(
FileInterceptor('file', { limits: { fileSize: MAX_IMPORT_ZIP_BYTES } }),
)
@ZodResponse({ status: 202, type: ColdImportUploadResponseDto })
@ApiOperation({
summary: '上传数据压缩包触发整包 cold-import',
description:
'zip 内含 manifest.tables 声明的数据文件(CSV/JSON);manifest+assemblers 取自 PAC 代码库不接受上传。' +
'校验通过后落后台队列,返回 jobId 供轮询进度。',
})
async coldImport(
@TenantScope() scope: TenantScopeContext,
@CurrentUser() user: AuthenticatedUser,
@UploadedFile() file: UploadedZip | undefined,
@Body('dryRun') dryRun?: string,
) {
if (!file?.buffer?.length) {
throw new BizError(ApiCode.CLIENT_BAD_REQUEST, '未收到上传文件(字段名应为 file)');
}
// multipart 文本字段是字符串;'true' → dry-run 预检(不落库)
return this.importUpload.prepareAndEnqueue(scope, file.buffer, user.sub, dryRun === 'true');
}
@Get('cold-import/:jobId')
@ZodResponse({ status: 200, type: ColdImportStatusDto })
@ApiOperation({ summary: '查询一次 cold-import 进度(前端轮询,含运行中百分比)' })
async coldImportStatus(@Param('jobId') jobId: string) {
return this.importUpload.getProgress(jobId);
}
}
apps/pac-service/src/modules/admin/import-upload.service.ts 0 → 100644
import { Injectable, Logger } from '@nestjs/common';
import * as fs from 'node:fs';
import * as path from 'node:path';
import { randomUUID } from 'node:crypto';
import * as yaml from 'js-yaml';
import * as yauzl from 'yauzl';
import { ApiCode } from '@pac/types';
import { BizError } from '../../common/errors/biz-error';
import { PrismaService } from '../../prisma/prisma.service';
import { QueueProducer } from '../../queues/queue-producer.service';
/**
* ImportUploadService — 网页上传压缩包 → 触发 cold-import 的编排层。
*
* 安全核心:manifest.yaml + assemblers/ **永远取自 PAC 代码库** `data/<host>/`,不接受上传。
* importDirectory 从 manifest 的 host_name 决定写哪个 host,若让上传者带 manifest 即可篡改
* 写入目标。故:
* - 按 scope.hostId 反查 host.name → 只认代码库 data/<name>/ 那份 manifest+assemblers
* - 从 zip 里**仅白名单提取** manifest.tables[].file 声明的文件名(basename 匹配,
* 落 join(tmpDir, basename) → 天然挡 zip-slip 路径穿越)
* - manifest.host_name 天然 == 反查的 name → 不可能写到别的 host
*
* 长任务不在 HTTP 里同步跑:落盘就绪后 enqueue cold-import 队列,返回 jobId 供前端轮询。
*/
@Injectable()
export class ImportUploadService {
private readonly logger = new Logger(ImportUploadService.name);
constructor(
private readonly prisma: PrismaService,
private readonly queue: QueueProducer,
) {}
/** 代码库 host 数据根(与 ColdImportService.resolveHostDir 同约定)。 */
private dataRoot(): string {
return process.env.PAC_INCREMENTAL_DATA_DIR ?? path.resolve(process.cwd(), 'data');
}
/** 临时解压工作根;跑完由 processor 清理。 */
private workRoot(): string {
return process.env.PAC_IMPORT_WORK_DIR ?? path.join(this.dataRoot(), '.imports');
}
/**
* 校验 + 落盘 + 入队。返回前端轮询用的 jobId。
* @param scope 调用者(宿主)scope,来自 token
* @param zipBuffer 上传的 zip 内容(multer memoryStorage buffer)
* @param userId 触发人(进 triggeredBy)
*/
async prepareAndEnqueue(
scope: { hostId: string; tenantId: string },
zipBuffer: Buffer,
userId: string,
dryRun = false,
): Promise<{ jobId: string }> {
// 上传内容先落盘成临时 zip(yauzl 从文件流式解压);finally 删除
await fs.promises.mkdir(this.workRoot(), { recursive: true });
const zipPath = path.join(this.workRoot(), `upload-${randomUUID()}.zip`);
await fs.promises.writeFile(zipPath, zipBuffer);
try {
const host = await this.prisma.host.findUnique({ where: { id: scope.hostId } });
if (!host) {
throw new BizError(ApiCode.CLIENT_BAD_REQUEST, `host 不存在: ${scope.hostId}`);
}
// 1. 代码库 data/<name>/ 必须已配置 manifest(否则该 host 不支持文件导入)
const srcDir = path.join(this.dataRoot(), host.name);
const manifestPath = path.join(srcDir, 'manifest.yaml');
if (!fs.existsSync(manifestPath)) {
throw new BizError(
ApiCode.CLIENT_BAD_REQUEST,
`该宿主未配置文件导入(缺 data/${host.name}/manifest.yaml)`,
);
}
// 2. 白名单 = manifest.tables[].file(都是根级 basename,如 customer_basic_info.csv)
const manifest = yaml.load(fs.readFileSync(manifestPath, 'utf-8')) as {
tables?: Array<{ file?: string; group?: string }>;
};
const whitelist = new Set(
(manifest.tables ?? [])
.map((t) => t.file)
.filter((f): f is string => !!f)
.map((f) => path.basename(f)),
);
if (whitelist.size === 0) {
throw new BizError(
ApiCode.CLIENT_BAD_REQUEST,
`manifest.tables 为空,无可导入文件声明`,
);
}
// 3. 临时目录做成"迷你 data 根"(与 data/ 同构):
// tmpRoot/<host>/ ← manifest + assemblers + 解压的数据文件
// tmpRoot/_shared/ ← 共享字典层副本
// assembler 的 dict_includes 用 ../../_shared/dict/ 相对引用(相对 assemblers/ 上两级),
// 必须保住这个相对位置,否则共享字典找不到(CLI dry-run 实测抓到)。
const tmpRoot = path.join(this.workRoot(), `import-${randomUUID()}`);
const runDir = path.join(tmpRoot, host.name);
await fs.promises.mkdir(runDir, { recursive: true });
await fs.promises.copyFile(manifestPath, path.join(runDir, 'manifest.yaml'));
const assemblersSrc = path.join(srcDir, 'assemblers');
if (fs.existsSync(assemblersSrc)) {
await fs.promises.cp(assemblersSrc, path.join(runDir, 'assemblers'), { recursive: true });
}
const sharedSrc = path.join(this.dataRoot(), '_shared');
if (fs.existsSync(sharedSrc)) {
await fs.promises.cp(sharedSrc, path.join(tmpRoot, '_shared'), { recursive: true });
}
// 4. 流式解压 —— 仅提取白名单 basename(挡 zip-slip)。
// yauzl 对损坏包 / 路径穿越条目会 fail-closed(整包拒绝)→ 转友好 400,不冒泡成 500。
let extracted: Set<string>;
try {
extracted = await extractWhitelisted(zipPath, runDir, whitelist);
} catch (err) {
await fs.promises.rm(tmpRoot, { recursive: true, force: true });
throw new BizError(
ApiCode.CLIENT_BAD_REQUEST,
`压缩包解析失败(可能损坏或含非法路径条目): ${err instanceof Error ? err.message : err}`,
);
}
// 5. 宽容 + 成组完整性校验(不再要求全表齐 —— PAC 各主体独立,装载层已宽容跳过):
// ① 非空:压缩包至少含一个可识别数据文件
// ② 成组:manifest 声明 group 的表,组内要么全在、要么整组不在;部分在 → 报错
// (半组产不出主体,如 treatment_plan 头+行缺一 → planned 产 0 行)
// ③ 独立表(无 group)缺失:放行,只少那部分 facts
if (extracted.size === 0) {
await fs.promises.rm(tmpRoot, { recursive: true, force: true });
throw new BizError(
ApiCode.CLIENT_BAD_REQUEST,
`压缩包不含任何可识别的数据文件(需匹配 manifest.tables 声明的文件名)`,
);
}
const incompleteGroups = findIncompleteGroups(manifest.tables ?? [], extracted);
if (incompleteGroups.length > 0) {
await fs.promises.rm(tmpRoot, { recursive: true, force: true });
throw new BizError(
ApiCode.CLIENT_VALIDATION_FAILED,
`成组主体不完整:${incompleteGroups
.map((g) => `"${g.group}" 缺 ${g.missing.join(',')}`)
.join(';')}(该组表须一起上传,或整组都不传)`,
{ incompleteGroups },
);
}
this.logger.log(
`import 就绪 host=${host.name} runDir=${runDir} 文件=${extracted.size}/${whitelist.size}` +
(extracted.size < whitelist.size
? `(缺 ${whitelist.size - extracted.size} 张独立表,宽容跳过)`
: ''),
);
// 6. 入队,返回 jobId。dir=runDir(跑),cleanupDir=tmpRoot(processor 跑完清整个迷你根)
const { jobId } = await this.queue.enqueueColdImport({
hostId: host.id,
tenantId: scope.tenantId,
hostName: host.name,
dir: runDir,
cleanupDir: tmpRoot,
dryRun,
triggeredBy: `upload:${userId}`,
});
return { jobId };
} finally {
// 上传的原始 zip 用完即删(已解压到 runDir);迷你根由 processor 跑完清理
await fs.promises.rm(zipPath, { force: true }).catch(() => undefined);
}
}
/**
* 查一次进度(前端轮询)。job active 时补查运行中的 sync_log.metadata.cohortDone/cohortTotal
* → 前端算真实百分比(importDirectory 每批实时写)。dryRun 不建 sync_log,只有 state。
*/
async getProgress(jobId: string): Promise<{
state: string;
result: unknown;
failedReason: string | null;
progress: { done: number; total: number } | null;
}> {
const job = await this.queue.getColdImportJob(jobId);
if (!job) return { state: 'not_found', result: null, failedReason: null, progress: null };
let progress: { done: number; total: number } | null = null;
if (job.state === 'active' && job.hostId) {
const log = await this.prisma.syncLog.findFirst({
where: { hostId: job.hostId, status: 'running' },
orderBy: { startedAt: 'desc' },
select: { metadata: true },
});
const m = log?.metadata as { cohortDone?: number; cohortTotal?: number } | null;
if (m?.cohortTotal) progress = { done: m.cohortDone ?? 0, total: m.cohortTotal };
}
return {
state: job.state,
result: job.returnValue,
failedReason: job.failedReason,
progress,
};
}
}
/**
* yauzl 流式解压,只提取 basename 命中白名单的条目。
* 只用 basename + join(destDir, ...) → 无法逃逸目标目录(zip-slip 防护)。
* 返回实际提取到的 basename 集合(供缺文件校验)。
* export 供单测(tests/cold-import-extract.spec.ts)。
*/
export function extractWhitelisted(
zipPath: string,
destDir: string,
whitelist: Set<string>,
): Promise<Set<string>> {
return new Promise((resolve, reject) => {
const extracted = new Set<string>();
yauzl.open(zipPath, { lazyEntries: true }, (err, zipfile) => {
if (err || !zipfile) return reject(err ?? new Error('无法打开压缩包'));
zipfile.on('error', reject);
zipfile.on('end', () => resolve(extracted));
zipfile.on('entry', (entry: YauzlEntry) => {
const base = path.basename(entry.fileName);
// 目录项 / 白名单外 → 跳过。
// (路径穿越 `../` 由 yauzl 自身 fail-closed 直接 emit error;此处 basename+join 是纵深防御)
if (entry.fileName.endsWith('/') || base.includes('..') || !whitelist.has(base)) {
return zipfile.readEntry();
}
zipfile.openReadStream(entry, (e, readStream) => {
if (e || !readStream) return reject(e ?? new Error('读取压缩条目失败'));
const ws = fs.createWriteStream(path.join(destDir, base));
readStream.on('error', reject);
ws.on('error', reject);
ws.on('close', () => {
extracted.add(base);
zipfile.readEntry();
});
readStream.pipe(ws);
});
});
zipfile.readEntry();
});
});
}
/**
* 成组完整性校验(通用,读宿主 manifest 的 group 声明):
* 返回"部分在但不全"的组(半组产不出主体 → 上传应拒)。组内全在或整组都不在 → 不返回(放行)。
* export 供单测(tests/cold-import-groups.spec.ts)。
*/
export function findIncompleteGroups(
tables: Array<{ file?: string; group?: string }>,
extracted: Set<string>,
): Array<{ group: string; present: string[]; missing: string[] }> {
const groups = new Map<string, string[]>(); // group → [basename...]
for (const t of tables) {
if (t.group && t.file) {
const arr = groups.get(t.group) ?? [];
arr.push(path.basename(t.file));
groups.set(t.group, arr);
}
}
return [...groups.entries()]
.map(([group, members]) => ({
group,
present: members.filter((f) => extracted.has(f)),
missing: members.filter((f) => !extracted.has(f)),
}))
.filter((g) => g.present.length > 0 && g.missing.length > 0);
}
// ── yauzl 最小类型(遵循仓库"不引 @types"的先例,如 assistant.controller UploadedAudio)──
interface YauzlEntry {
fileName: string;
}
apps/pac-service/src/modules/sync/cold-import/clickhouse-source.service.ts
......@@ -134,6 +134,35 @@ export class ClickHouseSourceService {
/// 用途:增量 fetched=0 时分辨「DW 真没新数据(静默)」vs「查询条件失效在空转(告警)」
/// —— 2026-06-10 游标 ISO 格式 bug 让增量空转三天才被人工发现,此探针让它当天就暴露。
/// 探针绝不能影响同步主流程:任何异常吞掉返回 {}。
/**
* 对账用:把资源查询原样包一层,统计**去重患者数**(patient_id + brand 元组 —— 集团内
* 同 patient_id 跨品牌是两个人,必须带 brand)。复用摄入查询本身 → 筛选口径零重复维护。
* 失败返回 null(对账降级,不阻断日报)。查询超时放宽到 120s(大表 count distinct)。
*/
async countDistinctPatients(source: ClickHouseSource, innerSql: string): Promise<number | null> {
let client: ReturnType<typeof createClient> | null = null;
try {
const conn = this.resolveConnection(source);
client = createClient({
url: conn.url,
database: conn.database,
username: conn.username,
password: conn.password,
request_timeout: 120_000,
compression: { response: true, request: false },
});
const q = `SELECT uniqExact(patient_id, brand) AS c FROM (\n${innerSql}\n)`;
const rs = await client.query({ query: q, format: 'JSONEachRow' });
const rows = (await rs.json()) as Array<{ c: string | number }>;
return Number(rows[0]?.c ?? 0);
} catch (e) {
this.logger.warn(`[reconcile] countDistinctPatients 失败: ${e instanceof Error ? e.message : String(e)}`);
return null;
} finally {
if (client) await client.close();
}
}
async probeNewRowCounts(
source: ClickHouseSource,
perQuery: IncrementalConfig['perQuery'],
......
apps/pac-service/src/modules/sync/cold-import/cold-import.service.ts
......@@ -570,6 +570,37 @@ export class ColdImportService {
}
/**
* 数据对账 —— DW 源侧「去重患者数」(设计见 /docs/data-reconciliation)。
*
* 复用摄入用的 sql_source 查询本身(同一道筛选闸)→ diff 只剩真漂移,零重复维护。
* 只算 DW 侧;PAC 侧计数 + diff + 渲染由调用方(日报)做(它持 prisma + 按 fact_type)。
*
* @param queryKeys 要对账的 sql_source.queries key(如 fact_emr_treatment_out / image_finding_rows)
* @returns { queryKey: 去重患者数 | null(查失败) };无 sql_source(file/push 宿主)→ 空 {}
*/
async reconcileDwPatientCounts(
hostName: string,
queryKeys: string[],
): Promise<Record<string, number | null>> {
const out: Record<string, number | null> = {};
let manifest: ColdImportManifest;
try {
manifest = this.readManifest(this.resolveHostDir(hostName));
} catch (e) {
this.logger.warn(`[reconcile] 读 manifest 失败 host=${hostName}: ${e instanceof Error ? e.message : String(e)}`);
return out;
}
const source = manifest.sql_source;
if (!source) return out; // file/push 宿主无直连源,不对账
for (const key of queryKeys) {
const sql = source.queries[key];
if (!sql) { out[key] = null; continue; }
out[key] = await this.chSource.countDistinctPatients(source, sql);
}
return out;
}
/**
* raw 层列漂移检测(form A):本次推送行的列集 vs 历史同源 rawPayload 的列集。
* - 历史有、本次没有 → column_removed(疑似宿主删列/改名)
* - 本次有、历史没有 → column_added(疑似宿主加列/改名)
......@@ -2150,16 +2181,28 @@ export class ColdImportService {
'manifest 缺数据源:既无 sql_source 也无 tables[](schema refine 本该拦住,兜底报错)',
);
}
// 宽容装载:声明表文件缺失 → warn + 跳过(不 throw)。
// PAC 各主体 assembler 独立:缺某表只少那部分 facts。下游已宽容——
// transform lookup/from 缺表当空表、project input 缺表跳过、assembler primary.table
// 缺表产 0 行。成组主体的完整性由上传层(ImportUploadService)按 manifest group 声明前置守卫,
// 不在此处判(此处也服务 CLI 全量导入,不宜硬拦)。
const tables: Record<string, unknown[]> = {};
const skipped: string[] = [];
for (const t of manifest.tables) {
const filePath = path.join(dir, t.file);
if (!fs.existsSync(filePath)) {
throw new Error(`Raw table file not found: ${filePath}`);
skipped.push(t.table);
continue;
}
const format = t.format ?? inferFileFormat(t.file);
tables[t.table] = this.readRows(filePath, format);
this.logger.log(` loaded raw table "${t.table}" (${tables[t.table]!.length} rows)`);
}
if (skipped.length > 0) {
this.logger.warn(
` ${skipped.length} 张声明表文件缺失,已跳过(依赖它们的 assembler/transform 将当空表处理): ${skipped.join(', ')}`,
);
}
return tables;
}
......
apps/pac-service/src/modules/sync/cold-import/manifest.schema.ts
......@@ -26,6 +26,12 @@ export const RawTableSchema = z.object({
file: z.string().min(1).describe('e.g. tb_encounter.csv / tb_patient.json'),
/// 文件格式(默认按扩展名推断)
format: z.enum(['csv', 'json']).optional(),
/// 成组主体标签(宿主特有,可选)。同 group 的表在**网页上传**时要么全在、要么整组跳过
/// —— 由 ImportUploadService 通用地读此声明做完整性校验。用于"半组产不出主体"的强依赖
/// (如 treatment_planned = customer_treat_plan[头] + _item[行],缺任一整个资源产 0 行)。
/// 不标 = 独立表:缺失仅少那部分 facts(装载层已宽容跳过)。
/// 注:CLI 全量导入不受此约束(只 warn 跳过);仅上传路径按 group 前置守卫。
group: z.string().min(1).optional(),
});
export type RawTable = z.infer<typeof RawTableSchema>;
......
apps/pac-service/src/modules/sync/transforms/operators/filter.op.ts
......@@ -10,6 +10,7 @@ import { isEmptyText, type Row } from '../row';
* - not_empty: true — 字段非 null/undefined/空字符串
* - in: [...] — 字段值在列表中
* - equals: x — 字段值等于 x
* - lt/lte/gt/gte: n — 数值比较(字段解析为数字;非数值/空 → 不通过)
*/
export function runFilter(op: FilterOp, rows: Row[]): { rows: Row[]; dropped: number } {
const out: Row[] = [];
......@@ -127,7 +128,26 @@ function passAll(row: Row, where: Record<string, WhereRule>): boolean {
if (!set.has(String(v))) return false;
} else if ('equals' in rule) {
if (String(v) !== String(rule.equals)) return false;
} else if ('lt' in rule) {
const n = toNumber(v);
if (n === null || n >= rule.lt) return false;
} else if ('lte' in rule) {
const n = toNumber(v);
if (n === null || n > rule.lte) return false;
} else if ('gt' in rule) {
const n = toNumber(v);
if (n === null || n <= rule.gt) return false;
} else if ('gte' in rule) {
const n = toNumber(v);
if (n === null || n < rule.gte) return false;
}
}
return true;
}
/// 数值解析:空/非数字 → null(数值比较一律不通过)。CSV 字段是字符串,金额如 "-30.00"。
function toNumber(v: unknown): number | null {
if (v === null || v === undefined || (typeof v === 'string' && v.trim() === '')) return null;
const n = Number(v);
return Number.isFinite(n) ? n : null;
}
apps/pac-service/src/modules/sync/transforms/transforms.schema.ts
......@@ -69,6 +69,8 @@ export type ProjectOp = z.infer<typeof ProjectOpSchema>;
* - `not_empty: true` — 字符串非空 / null 过滤
* - `in: [...]` — 值在列表中
* - `equals: x` — 值等于 x
* - `lt/lte/gt/gte: n` — 数值比较(字段解析为数字;非数值/空 → 不通过)。
* 让退费正负切分等"业务口径 WHERE"能在 PAC 侧做(宿主导原表,PAC transforms 过滤)。
*/
export const WhereRuleSchema = z.union([
z.object({ not_empty: z.literal(true) }),
......@@ -82,6 +84,11 @@ export const WhereRuleSchema = z.union([
z.object({ real_no_tooth: z.array(z.union([z.string(), z.number()])).min(1) }),
z.object({ in: z.array(z.union([z.string(), z.number()])).min(1) }),
z.object({ equals: z.union([z.string(), z.number()]) }),
// 数值比较:字段解析为数字后比较;非数值/空值一律不通过(退费 receivable_this < 0 等)。
z.object({ lt: z.number() }),
z.object({ lte: z.number() }),
z.object({ gt: z.number() }),
z.object({ gte: z.number() }),
]);
export type WhereRule = z.infer<typeof WhereRuleSchema>;
......
apps/pac-service/src/queues/daily-health-report.service.ts
import { Injectable, Logger } from '@nestjs/common';
import { Cron } from '@nestjs/schedule';
import { Prisma } from '@prisma/client';
import { PrismaService } from '../prisma/prisma.service';
import { AlertService, type AlertLevel } from '../common/alerting/alert.service';
import { ColdImportService } from '../modules/sync/cold-import/cold-import.service';
/**
* 数据对账 scope —— DW 源(sql_source.queries key)↔ PAC 侧计数口径。
* 复用摄入查询做源计数(同闸),PAC 侧按对应 fact_type / 表数去重患者。设计见 /docs/data-reconciliation。
*/
type PacCountKind =
| { kind: 'patient' }
| { kind: 'fact'; types: string[] }
| { kind: 'return_visit' }
| { kind: 'image_ai' };
const RECONCILE_SCOPES: Array<{ label: string; dwKey: string; pac: PacCountKind }> = [
{ label: '总量', dwKey: 'fact_client_out', pac: { kind: 'patient' } },
{ label: '治疗病历', dwKey: 'fact_emr_treatment_out', pac: { kind: 'fact', types: ['encounter_record', 'emr_record', 'diagnosis_record', 'treatment_record', 'recommendation_record'] } },
{ label: '预约', dwKey: 'fact_appointment_out', pac: { kind: 'fact', types: ['appointment_record'] } },
{ label: '结算', dwKey: 'fact_settlement_out', pac: { kind: 'fact', types: ['payment_record'] } },
{ label: '影像', dwKey: 'image_finding_rows', pac: { kind: 'image_ai' } },
{ label: '咨询', dwKey: 'fact_consult_out', pac: { kind: 'fact', types: ['consultation_record'] } },
{ label: '回访', dwKey: 'fact_returnvisit_out', pac: { kind: 'return_visit' } },
{ label: '转介', dwKey: 'fact_customer_referee_out', pac: { kind: 'fact', types: ['referral_record'] } },
];
/**
* DailyHealthReportService — 每日 09:00(沪)给企微推一份**简洁的线上健康日报**。
......@@ -21,6 +43,7 @@ export class DailyHealthReportService {
constructor(
private readonly prisma: PrismaService,
private readonly alerter: AlertService,
private readonly coldImport: ColdImportService,
) {}
@Cron(process.env.PAC_HEALTH_REPORT_CRON || '0 9 * * *', {
......@@ -68,6 +91,31 @@ export class DailyHealthReportService {
const lagHours = lastSync?.cursorAfter ? cursorLagHours(lastSync.cursorAfter, now) : null;
const lagTag = lagHours == null ? '⚪' : lagHours > errorH ? '🔴' : lagHours > warnH ? '🟡' : '🟢';
// ── 每轮增量明细(近 24h)──
// DW 现为 2h 刷新 → 每轮理应有数据;**空轮(fetched=0)大概率是异常**(DW 没刷 / 游标失效),
// 计入 warning。失败轮同样计入。逐轮列出让"每次 pull 的情况"一眼可见。
const rounds = await this.prisma.syncLog.findMany({
where: { resource: 'incremental_bundle', startedAt: { gte: since24h } },
orderBy: { startedAt: 'desc' },
select: {
startedAt: true, endedAt: true, status: true,
fetched: true, transactionsWritten: true, factsEmitted: true, duplicates: true, failed: true,
},
take: 20,
});
const roundEmpty = rounds.filter((r) => r.status === 'success' && r.fetched === 0).length;
const roundFailed = rounds.filter((r) => r.status !== 'success' && r.status !== 'running').length;
const roundLines = rounds.map((r) => {
const t = fmtClock(r.startedAt);
const mins = r.endedAt ? Math.round((+new Date(r.endedAt) - +new Date(r.startedAt)) / 60000) : null;
const dur = mins == null ? '' : mins >= 60 ? ` ${(mins / 60).toFixed(1)}h` : ` ${mins}m`;
if (r.status === 'running') return ` ${t} ⏳ 进行中`;
if (r.status !== 'success') return ` ${t} ❌ 失败${dur}`;
if (r.fetched === 0) return ` ${t} ⚠️ 空轮(0行)${dur}`;
return ` ${t} ✅ 拉${fmtNum(r.fetched)} tx${fmtNum(r.transactionsWritten)}/dup${fmtNum(r.duplicates)}/失${r.failed} f+${fmtNum(r.factsEmitted)}${dur}`;
});
const roundTag = roundFailed > 0 ? '🔴' : roundEmpty > 0 ? '🟡' : '🟢';
// ── 重算:画像 / 召回 最近一次 ──
const [lastPersona, lastPlanGen] = await Promise.all([
this.prisma.personaRecomputeLog.findFirst({ orderBy: { startedAt: 'desc' }, select: { startedAt: true, status: true } }),
......@@ -143,13 +191,23 @@ export class DailyHealthReportService {
if (fail7d > 0 && lastSync == null) level = 'critical';
if (lagHours != null && lagHours > warnH) level = 'warning';
if (lastPersona?.status === 'failed' || (lastPlanGen == null)) level = 'warning';
// DW 2h 刷新下,空轮/失败轮是异常信号
if (roundEmpty > 0) level = 'warning';
if (lagHours != null && lagHours > errorH) level = 'critical';
if (roundFailed > 0) level = 'critical';
// ── 数据对账(DW 源 vs PAC,去重患者数;设计见 /docs/data-reconciliation)──
// v1 观察模式:原样报 diff,不设基线、不驱动报告级别(diff 含已知差如瑞泰未全摄,
// 先看一两周摸清"正常差"再回填基线)。DW 不可达 → 该行降级 "—",不阻断日报。
const reconcileLines = await this.buildReconcile();
const dateStr = new Date(now).toLocaleDateString('zh-CN', { timeZone: 'Asia/Shanghai' }).replace(/\//g, '-');
const title = `每日健康报告 · ${dateStr}`;
const body = [
`📥 摄入 ${lagTag} 上次 ${fmtTime(lastSync?.startedAt)} · DW延迟 ${lagHours == null ? '—' : lagHours.toFixed(1) + 'h'} · 近7日失败 ${fail7d}`,
`📥 增量每轮(近24h · ${roundTag} ${rounds.length}轮${roundEmpty ? ` · 空${roundEmpty}` : ''}${roundFailed ? ` · 失败${roundFailed}` : ''})` +
(roundLines.length ? '\n' + roundLines.join('\n') : ' 无'),
`🔄 重算 画像 ${fmtTime(lastPersona?.startedAt)}${lastPersona?.status && lastPersona.status !== 'success' ? `(${lastPersona.status})` : ''} · 召回 ${fmtTime(lastPlanGen?.startedAt)}`,
`📈 数据(近24h↑) 患者 ${fmtNum(patientsTotal)}(+${patients24h}) · 事实 ${fmtWan(factsTotal)}(+${fmtNum(facts24h)})`,
`🖼 画像(24h) 重算 ${fmtNum(personaRuns24h)} 次(${personaRunsZh}) · 覆盖 ${fmtNum(personaPatients)}/${fmtNum(patientsTotal)}(${coveragePct}%)`,
......@@ -157,10 +215,66 @@ export class DailyHealthReportService {
`📋 召回池 待办 ${fmtNum(plansActive)} · 进行中 ${fmtNum(plansAssigned)}`,
`📞 客服执行(24h) ${execTotal} 次${execBreakdown ? ` · ${execBreakdown}` : ''}`,
`🤖 AI(24h) ${inv24h} 次 · ¥${cost24h.toFixed(2)}`,
`⚖️ 对账(去重患者·DW源vsPAC)${reconcileLines.length ? '\n' + reconcileLines.join('\n') : ' —'}`,
].join('\n');
return { level, title, body };
}
/**
* 数据对账 —— 逐 scope 比 DW 源 vs PAC 的去重患者数。
* DW 侧复用摄入 sql_source 查询(同闸口径);PAC 侧按对应 fact_type / 表数去重患者。
* v1 只渲染原始 diff(不设基线、不驱动级别),供观察。失败降级 "—",绝不阻断日报。
*/
private async buildReconcile(): Promise<string[]> {
try {
const host = await this.prisma.host.findFirst({ where: { name: 'jvs-dw' }, select: { id: true } });
if (!host) return [];
const dwCounts = await this.coldImport.reconcileDwPatientCounts('jvs-dw', RECONCILE_SCOPES.map((s) => s.dwKey));
if (Object.keys(dwCounts).length === 0) return []; // 无 sql_source(file/push 宿主)
const lines: string[] = [];
for (const s of RECONCILE_SCOPES) {
const dw = dwCounts[s.dwKey] ?? null;
const pac = await this.pacPatientCount(host.id, s.pac);
lines.push(fmtReconcileLine(s.label, dw, pac));
}
return lines;
} catch (err) {
this.logger.warn(`对账段生成失败(降级,不阻断日报): ${err instanceof Error ? err.message : String(err)}`);
return [];
}
}
/** PAC 侧去重患者数(当前版本:superseded_at IS NULL)。 */
private async pacPatientCount(hostId: string, pac: PacCountKind): Promise<number | null> {
try {
if (pac.kind === 'patient') {
return this.prisma.patient.count({ where: { hostId } });
}
if (pac.kind === 'return_visit') {
const r = await this.prisma.$queryRaw<Array<{ c: bigint }>>`
SELECT count(distinct patient_id) AS c FROM patient_return_visit WHERE host_id = ${hostId}::uuid`;
return Number(r[0]?.c ?? 0);
}
if (pac.kind === 'image_ai') {
// 影像 AI → diagnosis_record 且 content.code_source='image_ai'(image_finding 装配打的标)
const r = await this.prisma.$queryRaw<Array<{ c: bigint }>>`
SELECT count(distinct patient_id) AS c FROM patient_facts
WHERE host_id = ${hostId}::uuid AND superseded_at IS NULL
AND type = 'diagnosis_record' AND content->>'code_source' = 'image_ai'`;
return Number(r[0]?.c ?? 0);
}
// fact 类型:跨多类型取去重患者并集(不是各类型之和)
const r = await this.prisma.$queryRaw<Array<{ c: bigint }>>`
SELECT count(distinct patient_id) AS c FROM patient_facts
WHERE host_id = ${hostId}::uuid AND superseded_at IS NULL
AND type IN (${Prisma.join(pac.types)})`;
return Number(r[0]?.c ?? 0);
} catch (err) {
this.logger.warn(`PAC 计数失败(${JSON.stringify(pac)}): ${err instanceof Error ? err.message : String(err)}`);
return null;
}
}
}
/** cursor_after JSON({resource: "2026-06-21 19:30:00", ...})→ 距今小时数(取最大游标)。 */
......@@ -198,3 +312,18 @@ function fmtTime(d: Date | null | undefined): string {
if (!d) return '—';
return new Date(d).toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai', month: '2-digit', day: '2-digit', hour: '2-digit', minute: '2-digit', hour12: false });
}
/** 只要 HH:mm(每轮明细用,同日多轮不必重复日期)。 */
function fmtClock(d: Date | null | undefined): string {
if (!d) return '—';
return new Date(d).toLocaleTimeString('zh-CN', { timeZone: 'Asia/Shanghai', hour: '2-digit', minute: '2-digit', hour12: false });
}
/** 对账单行:DW / PAC / 差(+ = PAC 缺,源比 PAC 多)。v1 只报数不判级,大差多为已知基线(如瑞泰未全摄)。 */
function fmtReconcileLine(label: string, dw: number | null, pac: number | null): string {
if (dw == null || pac == null) return ` ${label} DW ${dw == null ? '—' : fmtNum(dw)} / PAC ${pac == null ? '—' : fmtNum(pac)}`;
const diff = dw - pac;
const pct = dw > 0 ? (diff / dw) * 100 : 0;
const sign = diff > 0 ? '+' : '';
return ` ${label} DW${fmtNum(dw)}/PAC${fmtNum(pac)} 差${sign}${fmtNum(diff)}(${pct.toFixed(0)}%)`;
}
apps/pac-service/src/queues/job-payloads.ts
......@@ -40,3 +40,18 @@ export interface ExecutionCallbackJob {
executionId: string;
triggeredBy: string;
}
export interface ColdImportJob {
hostId: string;
tenantId: string;
/** manifest.host_name;仅日志用(importDirectory 从 dir 内 manifest 自定位 host) */
hostName: string;
/** cold-import 的 --dir:迷你 data 根下的 <host>/ 子目录(含 manifest+assemblers+数据文件) */
dir: string;
/** 清理根:迷你 data 根(dir 的父,含 <host>/ 和 _shared/);worker 跑完整个删掉 */
cleanupDir: string;
/** true = dry-run 预检:跑完整校验+翻译但不落库,返回 totals(failed/mappingMiss)供确认 */
dryRun?: boolean;
/** `upload:<userId>` */
triggeredBy: string;
}
apps/pac-service/src/queues/processors/cold-import.processor.ts 0 → 100644
import { Logger } from '@nestjs/common';
import { Processor, WorkerHost } from '@nestjs/bullmq';
import { Job } from 'bullmq';
import * as fs from 'node:fs';
import { QueueName } from '../queue-names';
import type { ColdImportJob } from '../job-payloads';
import { ColdImportService } from '../../modules/sync/cold-import/cold-import.service';
/**
* cold-import worker — 网页上传压缩包触发的整包导入。
*
* 流程:
* 1. 调 ColdImportService.importDirectory(job.dir) — 与 CLI `pnpm cold-import` 同一入口;
* 内部自建 syncLog(审计/最近事件)、cohort 分批、幂等落库。返回 ImportRunResult。
* 2. finally 清理临时解压目录(含 PII 的上传数据文件不残留);失败时保留供排查。
*
* concurrency=1:整包导入是重任务,且 importDirectory 有 partial UNIQUE(host_id)
* WHERE status='running' 并发锁 —— 串行天然避免撞锁。attempts=1 不自动重试。
*
* 返回值即 job.returnvalue,前端按 jobId 轮询时读其中 totals 展示结果。
*/
@Processor(QueueName.COLD_IMPORT, { concurrency: 1 })
export class ColdImportProcessor extends WorkerHost {
private readonly logger = new Logger(ColdImportProcessor.name);
constructor(private readonly coldImport: ColdImportService) {
super();
}
async process(job: Job<ColdImportJob>) {
const { hostName, dir, cleanupDir, dryRun, triggeredBy } = job.data;
this.logger.log(
`cold-import START job=${job.id} host=${hostName} dir=${dir} dryRun=${!!dryRun} triggeredBy=${triggeredBy}`,
);
try {
const result = await this.coldImport.importDirectory(dir, { dryRun: !!dryRun });
this.logger.log(
`cold-import DONE job=${job.id} host=${hostName} status=${result.status} ` +
`patients=${result.totals.patientsUpserted} txn=${result.totals.transactionsWritten} ` +
`dup=${result.totals.duplicates} failed=${result.totals.failed} ` +
`factsCreated=${result.totals.factsCreated} factsSuperseded=${result.totals.factsSuperseded}`,
);
return result;
} finally {
// 成功/失败都清理整个迷你 data 根(cleanupDir,含 <host>/ 数据 + _shared/ 副本)。
// 始终清理(PII 优先),失败原因已在 syncLog.errorMessage + job.failedReason 里。
const target = cleanupDir ?? dir; // 兼容旧 job(无 cleanupDir)
try {
await fs.promises.rm(target, { recursive: true, force: true });
this.logger.debug(`cold-import 清理临时目录 job=${job.id} dir=${target}`);
} catch (err) {
this.logger.warn(
`cold-import 清理临时目录失败 job=${job.id} dir=${target}: ${err instanceof Error ? err.message : err}`,
);
}
}
}
}
apps/pac-service/src/queues/queue-names.ts
......@@ -15,6 +15,8 @@ export const QueueName = {
PLAN_ASSET_GENERATE: 'plan-asset-generate',
/// 执行结果回执投递(PAC → 宿主 callbackUrl,HMAC 签名 + 退避重试;jobName = executionId)
EXECUTION_CALLBACK: 'execution-callback',
/// 网页上传压缩包触发的整包 cold-import(长任务,jobName = jobId;concurrency=1 串行)
COLD_IMPORT: 'cold-import',
} as const;
export type QueueName = (typeof QueueName)[keyof typeof QueueName];
......@@ -53,4 +55,14 @@ export const QueueDefaults = {
removeOnComplete: { count: 500 },
removeOnFail: { count: 5_000 },
},
[QueueName.COLD_IMPORT]: {
/**
* 整包导入是几十分钟的重任务,且 importDirectory 有 partial UNIQUE(host_id) WHERE status='running'
* 并发锁(重试会撞锁)→ **不自动重试**(attempts:1),失败就留 failed 供人工排查。
* 完成的 job 保留久一点(前端靠 jobId 轮询进度,别在读到前被清)。
*/
attempts: 1,
removeOnComplete: { age: 24 * 3_600, count: 100 },
removeOnFail: { count: 1_000 },
},
} as const;
apps/pac-service/src/queues/queue-producer.service.ts
import { Injectable, Logger } from '@nestjs/common';
import { InjectQueue } from '@nestjs/bullmq';
import { randomUUID } from 'node:crypto';
import type { Queue } from 'bullmq';
import { QueueName, QueueDefaults } from './queue-names';
import type {
......@@ -7,6 +8,7 @@ import type {
PlanRecomputeJob,
PlanAssetGenerateJob,
ExecutionCallbackJob,
ColdImportJob,
} from './job-payloads';
/**
......@@ -29,6 +31,7 @@ export class QueueProducer {
@InjectQueue(QueueName.PLAN_RECOMPUTE) private readonly planQ: Queue,
@InjectQueue(QueueName.PLAN_ASSET_GENERATE) private readonly assetQ: Queue,
@InjectQueue(QueueName.EXECUTION_CALLBACK) private readonly callbackQ: Queue,
@InjectQueue(QueueName.COLD_IMPORT) private readonly coldImportQ: Queue,
) {}
async enqueuePersonaRecompute(
......@@ -90,6 +93,41 @@ export class QueueProducer {
);
}
/**
* 整包 cold-import 入队。返回 jobId 作前端轮询句柄(不落 syncLog —— importDirectory 自己会写
* 自己的 syncLog 供审计/最近事件;前端进度直接查 BullMQ job 的 state + returnvalue)。
* concurrency=1 串行 + importDirectory 自带 running 锁 → 同 host 不会并发跑。
*/
async enqueueColdImport(job: ColdImportJob): Promise<{ jobId: string }> {
const jobId = randomUUID();
await this.coldImportQ.add(job.hostName, job, {
jobId,
...QueueDefaults[QueueName.COLD_IMPORT],
});
this.logger.log(
`enqueue cold-import job=${jobId} host=${job.hostName} triggeredBy=${job.triggeredBy}`,
);
return { jobId };
}
/** 查一次 cold-import job 进度(前端轮询用)。job 不存在返回 null(可能已被清理)。 */
async getColdImportJob(jobId: string): Promise<{
state: string;
hostId: string | null;
returnValue: unknown;
failedReason: string | null;
} | null> {
const job = await this.coldImportQ.getJob(jobId);
if (!job) return null;
const state = await job.getState();
return {
state,
hostId: (job.data as { hostId?: string } | undefined)?.hostId ?? null,
returnValue: job.returnvalue ?? null,
failedReason: job.failedReason ?? null,
};
}
async enqueueExecutionCallback(job: ExecutionCallbackJob): Promise<void> {
// jobId = executionId(append-only 唯一)→ 同一 execution 不会重复入队
await this.callbackQ.add(job.executionId, job, {
......
apps/pac-service/src/queues/queues.module.ts
......@@ -17,6 +17,7 @@ import { PersonaRecomputeProcessor } from './processors/persona-recompute.proces
import { PlanRecomputeProcessor } from './processors/plan-recompute.processor';
import { PlanAssetGenerateProcessor } from './processors/plan-asset-generate.processor';
import { ExecutionCallbackProcessor } from './processors/execution-callback.processor';
import { ColdImportProcessor } from './processors/cold-import.processor';
/**
* Queues — BullMQ 三个队列的统一注册入口。
......@@ -56,6 +57,7 @@ import { ExecutionCallbackProcessor } from './processors/execution-callback.proc
{ name: QueueName.PLAN_RECOMPUTE },
{ name: QueueName.PLAN_ASSET_GENERATE },
{ name: QueueName.EXECUTION_CALLBACK },
{ name: QueueName.COLD_IMPORT },
),
PersonaModule,
PlanModule,
......@@ -73,6 +75,7 @@ import { ExecutionCallbackProcessor } from './processors/execution-callback.proc
PlanRecomputeProcessor,
PlanAssetGenerateProcessor,
ExecutionCallbackProcessor,
ColdImportProcessor,
],
exports: [BullModule, QueueProducer, StaleScanService, SyncIncrementalSchedulerService],
})
......
apps/pac-service/src/types/yauzl.d.ts 0 → 100644
// yauzl 无官方 @types —— 遵循仓库"不引额外类型包"的先例(见 assistant.controller、mcp-sdk.d.ts),
// 只声明本项目实际用到的最小 API 面。
declare module 'yauzl' {
import type { EventEmitter } from 'node:events';
import type { Readable } from 'node:stream';
export interface Entry {
fileName: string;
}
export interface ZipFile extends EventEmitter {
readEntry(): void;
openReadStream(
entry: Entry,
callback: (err: Error | null, stream?: Readable) => void,
): void;
}
export interface OpenOptions {
lazyEntries?: boolean;
}
export function open(
path: string,
options: OpenOptions,
callback: (err: Error | null, zipfile?: ZipFile) => void,
): void;
}
apps/pac-service/tests/cold-import-extract.spec.ts 0 → 100644
/**
* cold-import 上传解压 — extractWhitelisted 白名单 + zip-slip 防护
*
* 覆盖:
* - 白名单过滤:zip 里白名单外的文件(c.csv)不提取
* - 子目录条目按 basename 落平(sub/nested.csv → nested.csv,若在白名单)
* - 缺文件:zip 缺白名单声明的文件 → extracted 集合据此可算出缺哪些
* - zip-slip:含 `../evil.csv` 路径穿越条目的包被 yauzl fail-closed **整包拒绝**(reject)
*
* fixtures 由 tests/fixtures/cold-import/*.zip 提供(python3 zipfile 预造,含合法 arcname `../`)。
*
* 跑:pnpm --filter @pac/service test -- cold-import-extract
*/
import * as fs from 'node:fs';
import * as os from 'node:os';
import * as path from 'node:path';
import { extractWhitelisted } from '../src/modules/admin/import-upload.service';
const FIX = path.join(__dirname, 'fixtures', 'cold-import');
function tmpDir(): string {
return fs.mkdtempSync(path.join(os.tmpdir(), 'cold-import-test-'));
}
describe('extractWhitelisted', () => {
it('只提取白名单命中的文件,忽略白名单外条目', async () => {
const dest = tmpDir();
try {
const got = await extractWhitelisted(
path.join(FIX, 'good.zip'),
dest,
new Set(['a.csv', 'b.csv']),
);
expect([...got].sort()).toEqual(['a.csv', 'b.csv']);
expect(fs.existsSync(path.join(dest, 'a.csv'))).toBe(true);
expect(fs.existsSync(path.join(dest, 'b.csv'))).toBe(true);
} finally {
fs.rmSync(dest, { recursive: true, force: true });
}
});
it('缺文件:zip 缺声明文件 → extracted 不含它(供缺文件校验)', async () => {
const dest = tmpDir();
try {
const whitelist = new Set(['a.csv', 'b.csv']);
const got = await extractWhitelisted(path.join(FIX, 'partial.zip'), dest, whitelist);
const missing = [...whitelist].filter((f) => !got.has(f));
expect(missing).toEqual(['b.csv']);
} finally {
fs.rmSync(dest, { recursive: true, force: true });
}
});
it('zip-slip:含 `../` 路径穿越条目的包被整包拒绝,且父目录零写入', async () => {
const parent = tmpDir();
const dest = path.join(parent, 'inner');
fs.mkdirSync(dest);
try {
await expect(
extractWhitelisted(path.join(FIX, 'evil.zip'), dest, new Set(['a.csv', 'evil.csv'])),
).rejects.toThrow(/relative path|invalid/i);
// 关键:父目录绝无 evil.csv 写入(未逃逸)
expect(fs.existsSync(path.join(parent, 'evil.csv'))).toBe(false);
} finally {
fs.rmSync(parent, { recursive: true, force: true });
}
});
it('白名单外文件(c.csv)不落盘,子目录条目按 basename 落平', async () => {
const dest = tmpDir();
try {
const got = await extractWhitelisted(
path.join(FIX, 'extra.zip'),
dest,
new Set(['a.csv', 'nested.csv']),
);
expect([...got].sort()).toEqual(['a.csv', 'nested.csv']);
expect(fs.existsSync(path.join(dest, 'a.csv'))).toBe(true);
expect(fs.existsSync(path.join(dest, 'nested.csv'))).toBe(true); // sub/nested.csv → nested.csv
expect(fs.existsSync(path.join(dest, 'c.csv'))).toBe(false); // 白名单外
} finally {
fs.rmSync(dest, { recursive: true, force: true });
}
});
});
apps/pac-service/tests/cold-import-groups.spec.ts 0 → 100644
/**
* cold-import 成组完整性 — findIncompleteGroups
*
* 规则(用户定):同 manifest group 的表在网页上传时要么全在、要么整组不在;部分在 → 拒(半组无意义)。
* 独立表(无 group)缺失不影响(装载层宽容跳过)。
*
* 跑:pnpm --filter @pac/service test -- cold-import-groups
*/
import { findIncompleteGroups } from '../src/modules/admin/import-upload.service';
// 模拟 FRIDAY manifest.tables 的关键子集:treatment_plan 组(头+行)+ 若干独立表
const TABLES = [
{ file: 'customer_basic_info.csv' }, // 独立
{ file: 'appointment_base.csv' }, // 独立
{ file: 'customer_treat_plan.csv', group: 'treatment_plan' },
{ file: 'customer_treat_plan_item.csv', group: 'treatment_plan' },
];
describe('findIncompleteGroups', () => {
it('组内全在 → 不报(通过)', () => {
const got = findIncompleteGroups(
TABLES,
new Set(['customer_basic_info.csv', 'customer_treat_plan.csv', 'customer_treat_plan_item.csv']),
);
expect(got).toEqual([]);
});
it('整组都不在 → 不报(宽容:该主体不导)', () => {
const got = findIncompleteGroups(TABLES, new Set(['customer_basic_info.csv']));
expect(got).toEqual([]);
});
it('组内部分在(缺行表)→ 报不完整', () => {
const got = findIncompleteGroups(
TABLES,
new Set(['customer_basic_info.csv', 'customer_treat_plan.csv']), // 缺 _item
);
expect(got).toHaveLength(1);
expect(got[0]!.group).toBe('treatment_plan');
expect(got[0]!.missing).toEqual(['customer_treat_plan_item.csv']);
expect(got[0]!.present).toEqual(['customer_treat_plan.csv']);
});
it('独立表缺失不影响(无 group 不参与成组)', () => {
const got = findIncompleteGroups(
TABLES,
new Set(['customer_treat_plan.csv', 'customer_treat_plan_item.csv']), // 缺两张独立表,但组齐
);
expect(got).toEqual([]);
});
});
apps/pac-service/tests/filter-numeric.spec.ts 0 → 100644
/**
* filter 数值比较算子 — lt/lte/gt/gte
*
* 动机:让"业务口径 WHERE"(如退费正负切分 receivable_this < 0)能在 PAC transforms 做,
* 宿主只需导原表。非数值/空值一律不通过(空金额不算负数)。
*
* 跑:pnpm --filter @pac/service test -- filter-numeric
*/
import { runFilter } from '../src/modules/sync/transforms/operators/filter.op';
import type { FilterOp } from '../src/modules/sync/transforms/transforms.schema';
const rows = [
{ m: '-30.00' },
{ m: '0' },
{ m: '5' },
{ m: '' }, // 空
{ m: 'abc' }, // 非数值
{ m: null as unknown as string },
];
function filt(where: FilterOp['where']) {
const op = { kind: 'filter', input: 'x', output: 'y', where } as FilterOp;
return runFilter(op, rows).rows.map((r) => r.m);
}
describe('filter 数值算子', () => {
it('lt:< 0 只保留负数,空/非数值不通过', () => {
expect(filt({ m: { lt: 0 } })).toEqual(['-30.00']);
});
it('gte:>= 0 保留 0 和正数', () => {
expect(filt({ m: { gte: 0 } })).toEqual(['0', '5']);
});
it('gt:> 0 只保留正数', () => {
expect(filt({ m: { gt: 0 } })).toEqual(['5']);
});
it('lte:<= 0 保留 0 和负数', () => {
expect(filt({ m: { lte: 0 } })).toEqual(['-30.00', '0']);
});
it('空值 / 非数值 / null 一律不通过数值比较', () => {
// 用一个恒真下界也不会把空/非数值放进来
expect(filt({ m: { gte: -1e9 } })).toEqual(['-30.00', '0', '5']);
});
});
apps/pac-service/tests/friday-settlement-where.spec.ts 0 → 100644
/**
* FRIDAY 结算/病历 WHERE 迁移到 PAC transforms —— 验证"导原表 → PAC 侧 filter 切分"正确。
*
* 背景:业务 WHERE 从 export.sh 全部搬进 manifest transforms(单一真理源)。
* 本测用合成的**原表**数据(全 status + 正负金额)喂 TransformEngine,断言四处切分:
* ① 消费 status∈{1,3} 且 receivable≥0(负额被 gte 挡)
* ② 退费① status=4 ∪ (status=3 且 receivable<0) ← filter×2 + union
* ③ 退费明细 is_refund=1
* ④ 病历 status∈{3,4} 排草稿
*
* 跑:pnpm --filter @pac/service test -- friday-settlement-where
*/
import { TransformEngine } from '../src/modules/sync/transforms/transform-engine';
import type { TransformsConfig } from '../src/modules/sync/transforms/transforms.schema';
const engine = new TransformEngine();
// 合成原表(CSV 源天然全字符串;med_emr_info 是 JSON,status 用数字测 in 的 String 归一)
const patient_settlement = [
{ uuid: 's1', status: '1', receivable_this: '100', patient_id: '666716' }, // 消费
{ uuid: 's3', status: '3', receivable_this: '50', patient_id: '666716' }, // 消费(status3 正额)
{ uuid: 's1neg', status: '1', receivable_this: '-5', patient_id: '666716' }, // 消费但金额<0 → 被 gte 挡
{ uuid: 's4', status: '4', receivable_this: '-30', patient_id: '666716' }, // 退费(status4 整单反向)
{ uuid: 's3neg', status: '3', receivable_this: '-20', patient_id: '666716' }, // 退费(status3 负额表达)
{ uuid: 's2', status: '2', receivable_this: '10', patient_id: '666716' }, // 草稿/其他 → 都不收
];
// spec.patient_id 是「诊所本地垃圾 id」(元和王永品牌实证)→ 退费明细身份须从父结算头 lookup 纠正
const patient_settlement_spec = [
{ id: 'sp1', is_refund: '1', settlement_id: 's3', patient_id: '2' }, // 退费明细(本地 id 2,应被纠成 666716)
{ id: 'sp2', is_refund: '0', settlement_id: 's1', patient_id: '2' }, // 非退费
];
const med_emr_info = [
{ emr_sub_id: 'e3', status: 3 }, // 正式
{ emr_sub_id: 'e4', status: 4 }, // 正式
{ emr_sub_id: 'e2', status: 2 }, // 草稿 → 排除
];
const transforms = [
{ kind: 'filter', input: 'patient_settlement', output: '_pay_hdr', where: { status: { in: ['1', '3'] }, receivable_this: { gte: 0 } } },
{ kind: 'filter', input: 'patient_settlement', output: '_refund_status4', where: { status: { equals: '4' } } },
{ kind: 'filter', input: 'patient_settlement', output: '_refund_neg3', where: { status: { equals: '3' }, receivable_this: { lt: 0 } } },
{ kind: 'union', inputs: ['_refund_status4', '_refund_neg3'], output: 'refund_full_rows' },
{ kind: 'filter', input: 'patient_settlement_spec', output: '_refund_item_raw', where: { is_refund: { equals: '1' } } },
{ kind: 'lookup', input: '_refund_item_raw', output: 'refund_item_rows', from: 'patient_settlement', left_key: 'settlement_id', right_key: 'uuid', select: { owner_patient_id: 'patient_id' } },
{ kind: 'filter', input: 'med_emr_info', output: '_emr_official', where: { status: { in: ['3', '4'] } } },
] as unknown as TransformsConfig;
describe('FRIDAY 结算/病历 WHERE 迁移(导原表 + PAC 切分)', () => {
const out = engine.run({
tables: { patient_settlement, patient_settlement_spec, med_emr_info },
transforms,
});
const pick = (rows: unknown[], key: string) =>
(rows as Array<Record<string, unknown>>).map((r) => r[key]).sort();
it('① 消费:status∈{1,3} 且 receivable≥0(负额 s1neg 被挡,s2/s4 不在)', () => {
expect(pick(out['_pay_hdr'] ?? [], 'uuid')).toEqual(['s1', 's3']);
});
it('② 退费①:status=4 ∪ (status=3 且金额<0)', () => {
expect(pick(out['refund_full_rows'] ?? [], 'uuid')).toEqual(['s3neg', 's4']);
});
it('③ 退费明细:is_refund=1', () => {
expect(pick(out['refund_item_rows'] ?? [], 'id')).toEqual(['sp1']);
});
it('③.1 退费明细身份纠正:spec 本地 id 2 → 从父结算头继承真 patient_id 666716', () => {
const rows = (out['refund_item_rows'] ?? []) as Array<Record<string, unknown>>;
expect(rows.map((r) => r['owner_patient_id'])).toEqual(['666716']);
});
it('④ 病历:status∈{3,4} 排草稿 e2', () => {
expect(pick(out['_emr_official'] ?? [], 'emr_sub_id')).toEqual(['e3', 'e4']);
});
});
apps/pac-web/src/components/host-admin/cold-import-card.tsx 0 → 100644
'use client';
import { useEffect, useRef, useState } from 'react';
import { Upload, Loader2 } from 'lucide-react';
import { toast } from 'sonner';
import { Card, CardContent, CardDescription, CardHeader, CardTitle } from '@/components/ui/card';
import { Button } from '@/components/ui/button';
import { hostApi, type ColdImportStatus } from './host-api';
/** BullMQ 终态:轮询到这些状态就停。 */
const TERMINAL = new Set(['completed', 'failed', 'not_found']);
const POLL_MS = 3000;
const STATE_LABEL: Record<string, string> = {
waiting: '排队中',
delayed: '排队中',
active: '导入中',
completed: '完成',
failed: '失败',
not_found: '已结束',
};
function fmtElapsed(sec: number): string {
const m = Math.floor(sec / 60);
const s = sec % 60;
return m > 0 ? `${m}分${s}秒` : `${s}秒`;
}
/**
* ColdImportCard — 宿主自助上传数据压缩包触发 cold-import。
*
* 流程:选 zip → 上传拿 jobId → 每 3s 轮询进度 → 终态停 + 展示 totals + 刷新最近事件。
* manifest/assemblers 取自 PAC 代码库,上传包只含数据文件(见后端 ImportUploadService)。
*/
export function ColdImportCard({ onDone }: { onDone: () => void }) {
const inputRef = useRef<HTMLInputElement>(null);
const [file, setFile] = useState<File | null>(null);
const [dryRun, setDryRun] = useState(true); // 默认先预检,更安全
const [uploading, setUploading] = useState(false);
const [jobId, setJobId] = useState<string | null>(null);
const [status, setStatus] = useState<ColdImportStatus | null>(null);
const [startedAt, setStartedAt] = useState<number | null>(null);
const [elapsed, setElapsed] = useState(0);
// onDone 存 ref,避免父组件重渲染导致轮询 effect 被重置
const onDoneRef = useRef(onDone);
useEffect(() => {
onDoneRef.current = onDone;
}, [onDone]);
// 轮询:仅依赖 jobId
useEffect(() => {
if (!jobId) return;
let timer: ReturnType<typeof setInterval> | null = null;
let stopped = false;
const poll = async () => {
try {
const s = await hostApi.getImportStatus(jobId);
if (stopped) return;
setStatus(s);
if (TERMINAL.has(s.state)) {
if (timer) clearInterval(timer);
timer = null;
if (s.state === 'completed') {
if (s.result?.dryRun) {
// 预检:未落库,不刷新最近事件
toast.success('预检完成(未落库),检查下方 failed / 数量再正式导入');
} else {
toast.success('导入完成');
onDoneRef.current();
}
} else if (s.state === 'failed') {
toast.error(`导入失败:${s.failedReason ?? '未知原因'}`);
} else {
toast.message('任务已不在队列(可能已完成并被清理)');
}
}
} catch {
// 单次轮询失败不打断,等下一轮
}
};
void poll();
timer = setInterval(() => void poll(), POLL_MS);
return () => {
stopped = true;
if (timer) clearInterval(timer);
};
}, [jobId]);
const running = jobId != null && (!status || !TERMINAL.has(status.state));
// 已用时长:running 时每秒 tick
useEffect(() => {
if (!running || startedAt == null) return;
const t = setInterval(() => setElapsed(Math.floor((Date.now() - startedAt) / 1000)), 1000);
return () => clearInterval(t);
}, [running, startedAt]);
const onUpload = async () => {
if (!file) return;
setUploading(true);
setStatus(null);
setJobId(null);
try {
const form = new FormData();
form.append('file', file);
form.append('dryRun', String(dryRun));
const { jobId: id } = await hostApi.importArchive(form);
setJobId(id);
setStartedAt(Date.now());
setElapsed(0);
toast.success(dryRun ? '已上传,开始预检(不落库)' : '已上传,开始后台导入');
} catch (e) {
toast.error(e instanceof Error ? e.message : '上传失败');
} finally {
setUploading(false);
}
};
const t = status?.result?.totals;
return (
<Card>
<CardHeader>
<CardTitle>数据导入</CardTitle>
<CardDescription>
上传数据压缩包(zip,含 manifest 声明的 CSV/JSON)触发整包 cold-import。manifest / assemblers
取自 PAC 侧无需上传。重复上传同一批数据是安全的(幂等空转:去重高、fact 新建≈0)。
</CardDescription>
</CardHeader>
<CardContent className="space-y-3">
<input
ref={inputRef}
type="file"
accept=".zip"
className="hidden"
onChange={(e) => setFile(e.target.files?.[0] ?? null)}
/>
<div className="flex flex-wrap items-center gap-3">
<Button
variant="outline"
onClick={() => inputRef.current?.click()}
disabled={uploading || running}
>
选择 zip
</Button>
<span className="min-w-0 flex-1 truncate text-sm text-muted-foreground">
{file ? file.name : '未选择文件'}
</span>
<label className="flex items-center gap-1.5 text-sm text-muted-foreground">
<input
type="checkbox"
checked={dryRun}
onChange={(e) => setDryRun(e.target.checked)}
disabled={uploading || running}
/>
仅预检(不落库)
</label>
<Button onClick={onUpload} disabled={!file || uploading || running}>
{uploading ? (
<Loader2 className="mr-2 h-4 w-4 animate-spin" />
) : (
<Upload className="mr-2 h-4 w-4" />
)}
{dryRun ? '预检' : '上传并导入'}
</Button>
</div>
{jobId && (
<div className="rounded-md border p-3 text-sm">
<div className="flex items-center gap-2">
{running && <Loader2 className="h-4 w-4 animate-spin" />}
<span>
状态:{STATE_LABEL[status?.state ?? ''] ?? status?.state ?? '排队中'}
{status?.result?.dryRun ? '(预检)' : ''}
</span>
{(running || elapsed > 0) && (
<span className="text-xs text-muted-foreground">· 用时 {fmtElapsed(elapsed)}</span>
)}
</div>
{/* 运行中真实进度条(cohortDone/cohortTotal 批次) */}
{status?.progress && status.progress.total > 0 && (
<div className="mt-2">
<div className="h-1.5 w-full overflow-hidden rounded bg-muted">
<div
className="h-full bg-primary transition-all"
style={{
width: `${Math.round((status.progress.done / status.progress.total) * 100)}%`,
}}
/>
</div>
<div className="mt-1 text-xs text-muted-foreground">
已处理 {status.progress.done}/{status.progress.total} 批
</div>
</div>
)}
{/* 长任务提示:后台跑,可离开 */}
{running && (
<p className="mt-1.5 text-xs text-muted-foreground">
导入在后台进行,可离开此页,稍后回来查看结果。
</p>
)}
{status?.state === 'failed' && status.failedReason && (
<p className="mt-1 text-destructive">失败原因:{status.failedReason}</p>
)}
{t && (
<div className="mt-2 grid grid-cols-2 gap-x-4 gap-y-1 text-muted-foreground">
<span>患者 upsert:{t.patientsUpserted}</span>
<span>事务写入:{t.transactionsWritten}</span>
<span>幂等去重:{t.duplicates}</span>
<span>失败:{t.failed}</span>
<span>fact 新建:{t.factsCreated}</span>
<span>fact 更新:{t.factsSuperseded}</span>
<span>fact 未变:{t.factsUnchanged}</span>
<span>证据追加:{t.factsEvidenceAppended}</span>
</div>
)}
</div>
)}
</CardContent>
</Card>
);
}
apps/pac-web/src/components/host-admin/host-admin-app.tsx
......@@ -24,6 +24,7 @@ import { isPostMessageSentinel } from '@/lib/host-message';
import { hostApi } from './host-api';
import { useConfirm } from './use-confirm';
import { SecretRevealDialog } from './secret-reveal-dialog';
import { ColdImportCard } from './cold-import-card';
/**
* HostAdminApp — 宿主自己的 admin 配置页主视图。
......@@ -248,6 +249,9 @@ export function HostAdminApp({ data, onRefresh }: { data: HostDetail; onRefresh:
<StatsCard data={data} />
{/* 数据导入(上传压缩包触发 cold-import) */}
<ColdImportCard onDone={onRefresh} />
{/* 凭据 */}
<Card>
<CardHeader>
......
apps/pac-web/src/components/host-admin/host-api.ts
......@@ -13,6 +13,28 @@ import { api } from '@/lib/api-client';
const BASE = '/pac/v1/admin/host/self';
export interface ColdImportTotals {
patientsUpserted: number;
transactionsWritten: number;
duplicates: number;
failed: number;
factsCreated: number;
factsSuperseded: number;
factsUnchanged: number;
factsEvidenceAppended: number;
factsFailed: number;
}
export interface ColdImportStatus {
/** BullMQ job 状态:waiting/active/completed/failed/...;'not_found' = 已清理或不存在 */
state: string;
/** 完成时的 ImportRunResult(含 totals + dryRun 标志);进行中为 null */
result: { status?: string; dryRun?: boolean; totals?: ColdImportTotals } | null;
failedReason: string | null;
/** 运行中的真实进度(cohortDone/cohortTotal 批次);非运行态或 dryRun 为 null */
progress: { done: number; total: number } | null;
}
export const hostApi = {
getSelf: () => api.get<HostDetail>(BASE),
getStats: () => api.get<HostStats>(`${BASE}/stats`),
......@@ -28,4 +50,9 @@ export const hostApi = {
api.delete<{ totalKeysRemaining: number }>(
`${BASE}/callback-secrets/${encodeURIComponent(suffix)}`,
),
// 数据导入:上传压缩包(FormData 字段名 file)→ 返回 jobId;按 jobId 轮询进度
importArchive: (form: FormData) => api.post<{ jobId: string }>(`${BASE}/cold-import`, form),
getImportStatus: (jobId: string) =>
api.get<ColdImportStatus>(`${BASE}/cold-import/${encodeURIComponent(jobId)}`),
};
pnpm-lock.yaml
......@@ -235,6 +235,9 @@ importers:
ws:
specifier: ^8.21.0
version: 8.21.0
yauzl:
specifier: ^3.3.0
version: 3.3.0
zod:
specifier: ^4.4.3
version: 4.4.3
......
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