Commit 46fda545 by luoqi

fix(sync): map Friday dedicated customer service and isolate reparse sources

parent bcdd0f4c
Pipeline #3785 failed in 0 seconds
...@@ -5,7 +5,7 @@ icon: FileJson ...@@ -5,7 +5,7 @@ icon: FileJson
--- ---
> 本文是 [Push 通道](/docs/integration/channel-push)**形态 A** 在 FRIDAY SaaS 宿主的具体化:每个 `source` 的 JSON 字段清单。 > 本文是 [Push 通道](/docs/integration/channel-push)**形态 A** 在 FRIDAY SaaS 宿主的具体化:每个 `source` 的 JSON 字段清单。
> 字段形状与 PAC 已完成的存量摄入**完全一致**——宿主按此推送,与存量自动衔接: > 已接入字段与 PAC 存量摄入口径一致;标注「待接入」的扩展需双方完成联调后启用。宿主按契约推送,与存量自动衔接:
> 同行按幂等键去重,行更新(`updated_gmt_at` 变)自动版本演进。**重叠无害,宁多勿漏**。 > 同行按幂等键去重,行更新(`updated_gmt_at` 变)自动版本演进。**重叠无害,宁多勿漏**。
> 时间字段的可空语义与 `created_gmt_at` 兜底见 §1 通用约定。 > 时间字段的可空语义与 `created_gmt_at` 兜底见 §1 通用约定。
...@@ -64,9 +64,21 @@ icon: FileJson ...@@ -64,9 +64,21 @@ icon: FileJson
| `organization_id` | string | | 建档诊所 | | `organization_id` | string | | 建档诊所 |
| `contacts_tel` | string | | **默认联系号(宿主 inline)**——按上述规则挑出的有效号码(= customer_contacts 原列名) | | `contacts_tel` | string | | **默认联系号(宿主 inline)**——按上述规则挑出的有效号码(= customer_contacts 原列名) |
| `relationship` | string/number | | **该号归属(宿主 inline,= customer_contacts 原列名)**——`PhoneRelationshipEnum`:`1`本人 `2`爸爸 `3`妈妈 `4`爷爷 `5`奶奶 `6`朋友 `7`配偶 `8`子女 `9`其他(PAC 存 `isSelf`=码=='1') | | `relationship` | string/number | | **该号归属(宿主 inline,= customer_contacts 原列名)**——`PhoneRelationshipEnum`:`1`本人 `2`爸爸 `3`妈妈 `4`爷爷 `5`奶奶 `6`朋友 `7`配偶 `8`子女 `9`其他(PAC 存 `isSelf`=码=='1') |
| `follow_people_id` | string | | **专属客服用户 ID(宿主 inline,待接入)**,来自 `customer_follow_person.follow_people_id`,取值与变更要求见下节 |
| `follow_people` | string | | **专属客服姓名(宿主 inline,待接入)**,与 `follow_people_id` 来自同一条关系记录 |
| `has_active_complex_case` | boolean | | **是否有进行中的复杂病例(宿主 inline,新增)**——`1`/`true`=宿主正在跟进该患者。PAC 用作召回排除闸,见下方「跟进中患者不进召回池」 | | `has_active_complex_case` | boolean | | **是否有进行中的复杂病例(宿主 inline,新增)**——`1`/`true`=宿主正在跟进该患者。PAC 用作召回排除闸,见下方「跟进中患者不进召回池」 |
| `created_gmt_at` / `updated_gmt_at` | string(datetime) | ✅ | 建档/更新时间 | | `created_gmt_at` / `updated_gmt_at` | string(datetime) | ✅ | 建档/更新时间 |
#### 专属客服(待联调)
- 来源:`customer.customer_follow_person`,`is_default = 1`。
- 关联:`tenant_id` + `customer_id`;多门店存在不同客服时,先确认有效归属,不任取一条。
- 字段:`follow_people_id`(字符串用户 ID)、`follow_people`(姓名),来自同一条记录,inline 到 `customer_basic_info`。
- 变更:新增、换人、改名时重推患者主档;关系表单独变更也须触发,不能仅依赖主档更新时间。
- 取消:清空协议待联调;省略字段或传 `null` 不代表取消。
- 补推:接入完成后,按当前关系分批补推历史患者主档。
- 验收:首次分配、换人、取消、重复推送、跨品牌及多门店归属。
#### 跟进中患者不进召回池 —— `has_active_complex_case`(PAC 侧已就绪,待宿主推) #### 跟进中患者不进召回池 —— `has_active_complex_case`(PAC 侧已就绪,待宿主推)
**要解决的问题**:宿主自己正在跟进(已建复杂病例、客服/咨询师在推进)的患者,PAC 不该再发起召回 —— **要解决的问题**:宿主自己正在跟进(已建复杂病例、客服/咨询师在推进)的患者,PAC 不该再发起召回 ——
......
...@@ -19,6 +19,9 @@ field_mapping: ...@@ -19,6 +19,9 @@ field_mapping:
gender: sex # tinyint 1/2 → enum_mapping 归一;0/空 → 空 gender: sex # tinyint 1/2 → enum_mapping 归一;0/空 → 空
birthDate: birthday # date;空值入库为 null birthDate: birthday # date;空值入库为 null
medicalRecordNumber: file_number # 档案号(病历号,客服沟通用) medicalRecordNumber: file_number # 档案号(病历号,客服沟通用)
# 宿主 inline customer_follow_person(is_default=1),不是回访任务负责人。
dedicatedCsId: follow_people_id
dedicatedCsName: follow_people
# 宿主跟进闸(2026-07-30 新增)—— 宿主自己正在跟进的患者不进召回池。 # 宿主跟进闸(2026-07-30 新增)—— 宿主自己正在跟进的患者不进召回池。
# 宿主 inline:该患者名下**未删除**(complex_case_info.is_del = 1)且 case_stage ∈ # 宿主 inline:该患者名下**未删除**(complex_case_info.is_del = 1)且 case_stage ∈
# {待跟进, 已咨询, 已预约, 诊疗中} 的复杂病例是否存在(布尔,不要条数)。 # {待跟进, 已咨询, 已预约, 诊疗中} 的复杂病例是否存在(布尔,不要条数)。
......
...@@ -133,7 +133,13 @@ echo "── MySQL 导出 ──" ...@@ -133,7 +133,13 @@ echo "── MySQL 导出 ──"
# 患者主档 + inline 默认联系号(挑:非空号 → is_default↓ → contacts_type↑ → id↑;phone 与归属同源同一条) # 患者主档 + inline 默认联系号(挑:非空号 → is_default↓ → contacts_type↑ → id↑;phone 与归属同源同一条)
mysql_csv "SELECT cbi.id,cbi.name,cbi.sex,cbi.birthday,cbi.file_number,cbi.tenant_id,cbi.organization_id,cbi.created_gmt_at,cbi.updated_gmt_at, mysql_csv "SELECT cbi.id,cbi.name,cbi.sex,cbi.birthday,cbi.file_number,cbi.tenant_id,cbi.organization_id,cbi.created_gmt_at,cbi.updated_gmt_at,
(SELECT c.contacts_tel FROM customer.customer_contacts c WHERE c.customer_id=cbi.id AND c.contacts_tel<>'' ORDER BY c.is_default DESC,c.contacts_type ASC,c.id ASC LIMIT 1) AS contacts_tel, (SELECT c.contacts_tel FROM customer.customer_contacts c WHERE c.customer_id=cbi.id AND c.contacts_tel<>'' ORDER BY c.is_default DESC,c.contacts_type ASC,c.id ASC LIMIT 1) AS contacts_tel,
(SELECT c.relationship FROM customer.customer_contacts c WHERE c.customer_id=cbi.id AND c.contacts_tel<>'' ORDER BY c.is_default DESC,c.contacts_type ASC,c.id ASC LIMIT 1) AS relationship (SELECT c.relationship FROM customer.customer_contacts c WHERE c.customer_id=cbi.id AND c.contacts_tel<>'' ORDER BY c.is_default DESC,c.contacts_type ASC,c.id ASC LIMIT 1) AS relationship,
(SELECT f.follow_people_id FROM customer.customer_follow_person f WHERE f.tenant_id=cbi.tenant_id AND f.customer_id=cbi.id AND f.is_default=1
AND (SELECT COUNT(DISTINCT f2.follow_people_id) FROM customer.customer_follow_person f2 WHERE f2.tenant_id=cbi.tenant_id AND f2.customer_id=cbi.id AND f2.is_default=1)=1
ORDER BY f.updated_gmt_at DESC,f.id DESC LIMIT 1) AS follow_people_id,
(SELECT f.follow_people FROM customer.customer_follow_person f WHERE f.tenant_id=cbi.tenant_id AND f.customer_id=cbi.id AND f.is_default=1
AND (SELECT COUNT(DISTINCT f2.follow_people_id) FROM customer.customer_follow_person f2 WHERE f2.tenant_id=cbi.tenant_id AND f2.customer_id=cbi.id AND f2.is_default=1)=1
ORDER BY f.updated_gmt_at DESC,f.id DESC LIMIT 1) AS follow_people
FROM customer.customer_basic_info cbi $(pf cbi.id)" customer_basic_info.csv FROM customer.customer_basic_info cbi $(pf cbi.id)" customer_basic_info.csv
# 预约(全状态;transforms 丢草稿 10) # 预约(全状态;transforms 丢草稿 10)
mysql_csv "SELECT id,patient_appointment_id,organization_id,tenant_id,appointment_date,appointment_start,appointment_status,doctor_user_id,appointment_time_length,in_time,created_gmt_at,updated_gmt_at FROM \`arrail-appointment-server\`.appointment_base $(pf patient_appointment_id)" appointment_base.csv mysql_csv "SELECT id,patient_appointment_id,organization_id,tenant_id,appointment_date,appointment_start,appointment_status,doctor_user_id,appointment_time_length,in_time,created_gmt_at,updated_gmt_at FROM \`arrail-appointment-server\`.appointment_base $(pf patient_appointment_id)" appointment_base.csv
......
interface CliArgs {
dir: string;
host: string;
subjectTypes?: string[];
patientIds?: string[];
patientsFile?: string;
dryRun: boolean;
recompute: boolean;
help: boolean;
}
export function parseReparseArgs(argv: string[]): CliArgs {
const args: CliArgs = { dir: '', host: 'jvs-dw', dryRun: false, recompute: true, help: false };
for (const a of argv) {
if (a === '--help' || a === '-h') args.help = true;
else if (a === '--dry-run') args.dryRun = true;
else if (a === '--no-recompute') args.recompute = false;
else if (a.startsWith('--dir=')) args.dir = a.slice('--dir='.length);
else if (a.startsWith('--host=')) args.host = a.slice('--host='.length);
else if (a.startsWith('--subject-type=')) args.subjectTypes = a.slice('--subject-type='.length).split(',').map((s) => s.trim()).filter(Boolean);
else if (a.startsWith('--patient=')) args.patientIds = a.slice('--patient='.length).split(',').map((s) => s.trim()).filter(Boolean);
else if (a.startsWith('--patients-file=')) args.patientsFile = a.slice('--patients-file='.length).trim();
}
if (!args.dir) args.dir = `./data/${args.host}`;
return args;
}
...@@ -6,7 +6,7 @@ ...@@ -6,7 +6,7 @@
* *
* Usage: * Usage:
* pnpm reparse -- --host=jvs-dw --subject-type=diagnosis # 实跑 + 定向重算 * pnpm reparse -- --host=jvs-dw --subject-type=diagnosis # 实跑 + 定向重算
* pnpm reparse -- --host=jvs-dw --subject-type=diagnosis --dry-run # 只报会改多少,不写 * pnpm reparse -- --host=jvs-dw --subject-type=diagnosis --dry-run # 只统计事务范围,不比较字段、不写
* pnpm reparse -- --host=jvs-dw --patient=<uuid>,<uuid> # 限定患者(少量,命令行) * pnpm reparse -- --host=jvs-dw --patient=<uuid>,<uuid> # 限定患者(少量,命令行)
* pnpm reparse -- --host=jvs-dw --patients-file=./ids.txt # 限定患者(大批,见下) * pnpm reparse -- --host=jvs-dw --patients-file=./ids.txt # 限定患者(大批,见下)
* pnpm reparse -- --host=jvs-dw --subject-type=diagnosis --no-recompute # 只重衍 fact,稍后单独 recompute * pnpm reparse -- --host=jvs-dw --subject-type=diagnosis --no-recompute # 只重衍 fact,稍后单独 recompute
...@@ -40,35 +40,11 @@ import { PersonaService } from '../modules/persona/persona.service'; ...@@ -40,35 +40,11 @@ import { PersonaService } from '../modules/persona/persona.service';
import { PlanEngineService } from '../modules/plan/engine/plan-engine.service'; import { PlanEngineService } from '../modules/plan/engine/plan-engine.service';
import { disableSchedulersForCli } from './bootstrap-flags'; import { disableSchedulersForCli } from './bootstrap-flags';
interface CliArgs { import { parseReparseArgs } from './reparse-args';
dir: string;
host: string;
subjectTypes?: string[];
patientIds?: string[];
patientsFile?: string;
dryRun: boolean;
recompute: boolean;
help: boolean;
}
function parseArgs(argv: string[]): CliArgs {
const args: CliArgs = { dir: './data/jvs-dw', host: 'jvs-dw', dryRun: false, recompute: true, help: false };
for (const a of argv) {
if (a === '--help' || a === '-h') args.help = true;
else if (a === '--dry-run') args.dryRun = true;
else if (a === '--no-recompute') args.recompute = false;
else if (a.startsWith('--dir=')) args.dir = a.slice('--dir='.length);
else if (a.startsWith('--host=')) args.host = a.slice('--host='.length);
else if (a.startsWith('--subject-type=')) args.subjectTypes = a.slice('--subject-type='.length).split(',').map((s) => s.trim()).filter(Boolean);
else if (a.startsWith('--patient=')) args.patientIds = a.slice('--patient='.length).split(',').map((s) => s.trim()).filter(Boolean);
else if (a.startsWith('--patients-file=')) args.patientsFile = a.slice('--patients-file='.length).trim();
}
return args;
}
async function main(): Promise<void> { async function main(): Promise<void> {
const logger = new Logger('reparse-cli'); const logger = new Logger('reparse-cli');
const args = parseArgs(process.argv.slice(2)); const args = parseReparseArgs(process.argv.slice(2));
if (args.help) { if (args.help) {
// eslint-disable-next-line no-console // eslint-disable-next-line no-console
console.log( console.log(
......
...@@ -93,7 +93,7 @@ import { ...@@ -93,7 +93,7 @@ import {
type TenantResolver, type TenantResolver,
type SourceUnitResolver, type SourceUnitResolver,
} from './tenant-resolver'; } from './tenant-resolver';
import { mergePatientPreferences } from '../pipeline/patient-preferences.util'; import { mergeStoredPatientPreferences } from '../pipeline/patient-preferences.util';
import { buildPatientUpsertData, buildProfileUpsertData } from './patient-upsert.util'; import { buildPatientUpsertData, buildProfileUpsertData } from './patient-upsert.util';
/** 患者索引 key:集团内 external_id 可跨 source_unit 撞号,索引按 (source_unit, external_id) 建键。 */ /** 患者索引 key:集团内 external_id 可跨 source_unit 撞号,索引按 (source_unit, external_id) 建键。 */
...@@ -153,7 +153,7 @@ export class ColdImportService { ...@@ -153,7 +153,7 @@ export class ColdImportService {
* 覆盖:field/enum/keyword_mapping + transforms 算子 + parser 改动(长尾字典修补 99%)。 * 覆盖:field/enum/keyword_mapping + transforms 算子 + parser 改动(长尾字典修补 99%)。
* 不覆盖:改了 CH `sql_source` SELECT 本身(要从 DW 多拉原来没拉的列/表)→ 那才需真重摄。 * 不覆盖:改了 CH `sql_source` SELECT 本身(要从 DW 多拉原来没拉的列/表)→ 那才需真重摄。
* *
* @returns 受影响 patientId(调用方据此重算 persona/plan);dryRun 时只 diff 不写。 * @returns 受影响 patientId(调用方据此重算 persona/plan);dryRun 时只统计事务范围,不装配、不比较字段、不写入。
*/ */
async reparseFromTransactions(opts: { async reparseFromTransactions(opts: {
dir: string; dir: string;
...@@ -168,6 +168,9 @@ export class ColdImportService { ...@@ -168,6 +168,9 @@ export class ColdImportService {
}> { }> {
const absDir = path.resolve(opts.dir); const absDir = path.resolve(opts.dir);
const manifest = this.readManifest(absDir); const manifest = this.readManifest(absDir);
if (manifest.host_name !== opts.hostName) {
throw new Error(`reparse: manifest host=${manifest.host_name} 与目标 host=${opts.hostName} 不一致`);
}
const host = await this.prisma.host.findFirst({ where: { name: opts.hostName } }); const host = await this.prisma.host.findFirst({ where: { name: opts.hostName } });
if (!host) throw new Error(`reparse: host '${opts.hostName}' not found`); if (!host) throw new Error(`reparse: host '${opts.hostName}' not found`);
const tenantResolver = buildTenantResolver(manifest); const tenantResolver = buildTenantResolver(manifest);
...@@ -184,6 +187,21 @@ export class ColdImportService { ...@@ -184,6 +187,21 @@ export class ColdImportService {
); );
} }
const transforms = manifest.transforms ?? []; const transforms = manifest.transforms ?? [];
// 同一 subjectType 可来自不同源表(如 Friday 的本次治疗与计划)。按 action 路由,
// 不把计划行当成病历原文;action 仍无法区分的配置在写入前拒绝。
const routeActions = (cfg: AssemblerConfig): string[] | undefined => {
const peers = subjectCfgs.filter((c) => c.emits!.subjectType === cfg.emits!.subjectType);
const raw = traceRawSourceTable(cfg.primary.table, transforms);
const others = peers.filter((c) => traceRawSourceTable(c.primary.table, transforms) !== raw);
if (!others.length) return undefined;
const actions = [...new Set(peers.filter((c) => traceRawSourceTable(c.primary.table, transforms) === raw)
.map((c) => c.emits!.action))];
if (others.some((c) => actions.includes(c.emits!.action))) {
throw new Error(`reparse: ${cfg.canonical} 多个源表共享 action,无法安全区分原始数据`);
}
return actions;
};
const routes = new Map(targetCfgs.map((cfg) => [cfg, routeActions(cfg)]));
// 只能离线 reparse「transform 产出的 primary 表」:其 rawPayload 是 transform 的源行,可重放。 // 只能离线 reparse「transform 产出的 primary 表」:其 rawPayload 是 transform 的源行,可重放。
// 非 transform 产出的(如 image_finding_rows = CH SQL 视图)→ rawPayload 不是其装配输入 → 跳过(需走 DW)。 // 非 transform 产出的(如 image_finding_rows = CH SQL 视图)→ rawPayload 不是其装配输入 → 跳过(需走 DW)。
const reparseableCfgs = targetCfgs.filter((cfg) => { const reparseableCfgs = targetCfgs.filter((cfg) => {
...@@ -234,7 +252,7 @@ export class ColdImportService { ...@@ -234,7 +252,7 @@ export class ColdImportService {
: [null]; : [null];
for (const chunk of chunks) { for (const chunk of chunks) {
n += await this.prisma.patientTransaction.count({ n += await this.prisma.patientTransaction.count({
where: { hostId: host.id, subjectType: cfg.emits!.subjectType, ...(chunk ? { patientId: { in: chunk } } : {}) }, where: { hostId: host.id, subjectType: cfg.emits!.subjectType, ...(routes.get(cfg) ? { action: { in: routes.get(cfg)! } } : {}), ...(chunk ? { patientId: { in: chunk } } : {}) },
}); });
} }
this.logger.log(`reparse[dry]: ${cfg.canonical}(${cfg.emits!.subjectType}) txns=${n} → 实跑按版本流 supersede 变更的、跳过不变的`); this.logger.log(`reparse[dry]: ${cfg.canonical}(${cfg.emits!.subjectType}) txns=${n} → 实跑按版本流 supersede 变更的、跳过不变的`);
...@@ -261,7 +279,7 @@ export class ColdImportService { ...@@ -261,7 +279,7 @@ export class ColdImportService {
const seen = seenHashByTable.get(rawTable) ?? new Set<string>(); const seen = seenHashByTable.get(rawTable) ?? new Set<string>();
seenHashByTable.set(rawTable, seen); seenHashByTable.set(rawTable, seen);
const txs = await this.prisma.patientTransaction.findMany({ const txs = await this.prisma.patientTransaction.findMany({
where: { hostId: host.id, subjectType: cfg.emits!.subjectType, patientId: { in: batch } }, where: { hostId: host.id, subjectType: cfg.emits!.subjectType, ...(routes.get(cfg) ? { action: { in: routes.get(cfg)! } } : {}), patientId: { in: batch } },
select: { rawPayload: true, payloadHash: true }, select: { rawPayload: true, payloadHash: true },
}); });
for (const t of txs) { for (const t of txs) {
...@@ -1701,10 +1719,11 @@ export class ColdImportService { ...@@ -1701,10 +1719,11 @@ export class ColdImportService {
hostId_tenantId_sourceUnit_externalId: { hostId, tenantId, sourceUnit, externalId }, hostId_tenantId_sourceUnit_externalId: { hostId, tenantId, sourceUnit, externalId },
}, },
create: { hostId, tenantId, sourceUnit, externalId, ...patientData }, create: { hostId, tenantId, sourceUnit, externalId, ...patientData },
update: patientData, update: { ...patientData, preferences: undefined },
}), }),
'patient upsert', 'patient upsert',
); );
await mergeStoredPatientPreferences(this.prisma, patient.id, patientData.preferences as Prisma.InputJsonValue | undefined);
// 副表 1:1 upsert // 副表 1:1 upsert
await this.withDbRetry( await this.withDbRetry(
() => () =>
......
import type { Prisma } from '@prisma/client'; import { Prisma } from '@prisma/client';
/** /**
* 把 canonical patient row 的派生属性并入 patients.preferences(Json)——零 schema 迁移。 * 把 canonical patient row 的派生属性并入 patients.preferences(Json)——零 schema 迁移。
...@@ -42,3 +42,18 @@ export function mergePatientPreferences( ...@@ -42,3 +42,18 @@ export function mergePatientPreferences(
return Object.keys(base).length ? (base as Prisma.InputJsonValue) : undefined; return Object.keys(base).length ? (base as Prisma.InputJsonValue) : undefined;
} }
/** JSON 列按已提供的偏好合并,避免只推联系号时清掉专属客服,反之亦然。 */
export async function mergeStoredPatientPreferences(
db: { $executeRaw: (query: Prisma.Sql) => Promise<number> },
patientId: string,
preferences: Prisma.InputJsonValue | undefined,
): Promise<void> {
if (preferences === undefined) return;
await db.$executeRaw(Prisma.sql`
UPDATE patients SET preferences =
(CASE WHEN jsonb_typeof(preferences) = 'object' THEN preferences ELSE '{}'::jsonb END)
|| ${JSON.stringify(preferences)}::jsonb
WHERE id = ${patientId}::uuid
`);
}
...@@ -6,7 +6,7 @@ import { QueueProducer } from '../../../queues/queue-producer.service'; ...@@ -6,7 +6,7 @@ import { QueueProducer } from '../../../queues/queue-producer.service';
import { TransactionSynthesizer } from './transaction-synthesizer'; import { TransactionSynthesizer } from './transaction-synthesizer';
import { ParserPipeline } from './parser-pipeline.service'; import { ParserPipeline } from './parser-pipeline.service';
import type { EmitsConfig } from '../assembler/assembler.schema'; import type { EmitsConfig } from '../assembler/assembler.schema';
import { mergePatientPreferences } from './patient-preferences.util'; import { mergePatientPreferences, mergeStoredPatientPreferences } from './patient-preferences.util';
/** 患者索引 key:集团内 external_id 跨 source_unit 撞号,按 (source_unit, external_id) 建键。 */ /** 患者索引 key:集团内 external_id 跨 source_unit 撞号,按 (source_unit, external_id) 建键。 */
function patientIndexKey(sourceUnit: string, externalId: string): string { function patientIndexKey(sourceUnit: string, externalId: string): string {
...@@ -328,8 +328,9 @@ export class PipelineDispatcher { ...@@ -328,8 +328,9 @@ export class PipelineDispatcher {
const patient = await this.prisma.patient.upsert({ const patient = await this.prisma.patient.upsert({
where: { hostId_tenantId_sourceUnit_externalId: { hostId, tenantId, sourceUnit, externalId } }, where: { hostId_tenantId_sourceUnit_externalId: { hostId, tenantId, sourceUnit, externalId } },
create: { hostId, tenantId, sourceUnit, externalId, ...patientData }, create: { hostId, tenantId, sourceUnit, externalId, ...patientData },
update: patientData, update: { ...patientData, preferences: undefined },
}); });
await mergeStoredPatientPreferences(this.prisma, patient.id, patientData.preferences);
await this.prisma.patientProfile.upsert({ await this.prisma.patientProfile.upsert({
where: { patientId: patient.id }, where: { patientId: patient.id },
create: { patientId: patient.id, ...profileData }, create: { patientId: patient.id, ...profileData },
......
import { readFileSync } from 'node:fs';
import { join } from 'node:path';
import * as yaml from 'js-yaml';
import { AssemblerEngine } from '../src/modules/sync/assembler/assembler-engine';
import { AssemblerConfigSchema } from '../src/modules/sync/assembler/assembler.schema';
import { buildPatientUpsertData } from '../src/modules/sync/cold-import/patient-upsert.util';
import { ColdImportService } from '../src/modules/sync/cold-import/cold-import.service';
test.each([
['friday', 'customer_basic_info', { follow_people_id: 71, follow_people: '客服甲' }],
['jvs-dw', 'patient_full', { current_task_director_id: 71, current_task_director: '客服甲' }],
])('%s 专属客服经过真实 assembler 落到统一偏好', (host, _table, fields) => {
const config = AssemblerConfigSchema.parse(yaml.load(readFileSync(join(__dirname, '../data', host, 'assemblers/patient.yaml'), 'utf8')));
const result = new AssemblerEngine().assemble({ config, hostId: 'host', normalize: { amountUnit: 'yuan', timezone: 'Asia/Shanghai' }, tables: {
[config.primary.table]: [{ id: 'P1', patient_id: 'P1', name: '测试患者', ...fields }],
} });
expect(result.errors).toEqual([]);
expect(result.rows).toHaveLength(1);
const data = buildPatientUpsertData(result.rows[0]!.canonical as Record<string, unknown>, 'P1', { partial: true });
expect(data.preferences).toMatchObject({ dedicatedCs: { id: '71', name: '客服甲' } });
});
test('reparse 拒绝将 Friday 患者套用 5i5ya 配置,查询数据库前失败', async () => {
const hostQuery = jest.fn();
const svc = new ColdImportService({ host: { findFirst: hostQuery } } as never, null as never, null as never, null as never, null as never, null as never, null as never);
await expect(svc.reparseFromTransactions({ dir: join(__dirname, '../data/jvs-dw'), hostName: 'friday', dryRun: true })).rejects.toThrow('不一致');
expect(hostQuery).not.toHaveBeenCalled();
});
import { execFileSync } from 'node:child_process';
import { Prisma } from '@prisma/client';
import { mergePatientPreferences, mergeStoredPatientPreferences } from '../src/modules/sync/pipeline/patient-preferences.util';
// 显式指定本地容器;临时表遮蔽真实表,事务回滚,不读写患者业务数据。
const container = process.env.PAC_TEST_POSTGRES_CONTAINER;
(container ? describe : describe.skip)('preferences PostgreSQL 合并', () => {
it('客服与联系号分别更新仍保留对方;空推送不覆盖,换客服替换旧值', async () => {
const statements: string[] = [];
const db = { $executeRaw: async (q: Prisma.Sql) => {
let i = 0;
statements.push(q.sql.replace(/\?/g, () => "'" + String(q.values[i++]).replace(/'/g, "''") + "'") + ';');
return 1;
} };
const id = '00000000-0000-4000-8000-000000000001';
await mergeStoredPatientPreferences(db, id, mergePatientPreferences({ dedicatedCsId: 71, dedicatedCsName: '客服甲' }));
await mergeStoredPatientPreferences(db, id, mergePatientPreferences({ phoneRelationshipCode: '2', phoneRelationship: 'father' }));
await mergeStoredPatientPreferences(db, id, mergePatientPreferences({}));
await mergeStoredPatientPreferences(db, id, mergePatientPreferences({ dedicatedCsId: 72, dedicatedCsName: '客服乙' }));
const result = execFileSync('docker', ['exec', '-i', container!, 'psql', '-X', '-qAt', '-v', 'ON_ERROR_STOP=1', '-U', 'pac', '-d', 'pac'], {
input: `BEGIN; CREATE TEMP TABLE patients(id uuid, preferences jsonb); INSERT INTO patients VALUES ('${id}', '{"other":true}');\n${statements.join('\n')}\nSELECT preferences FROM patients; ROLLBACK;`, encoding: 'utf8', timeout: 15000,
});
expect(JSON.parse(result.trim())).toEqual({ other: true, dedicatedCs: { id: '72', name: '客服乙' },
contactPhone: { relationshipCode: '2', relationship: 'father', isSelf: false } });
});
});
import { join } from 'node:path';
import { ColdImportService } from '../src/modules/sync/cold-import/cold-import.service';
import { parseReparseArgs } from '../src/cli/reparse-args';
it('配置目录跟随 host,显式目录保留', () => {
expect(parseReparseArgs(['--host=friday']).dir).toBe('./data/friday');
expect(parseReparseArgs([]).dir).toBe('./data/jvs-dw');
expect(parseReparseArgs(['--dir=/custom', '--host=friday']).dir).toBe('/custom');
});
it('Friday 本次治疗/复查与治疗计划按 action 还原各自原表', async () => {
const actual = { emr_sub_id: 'E1', patient_id: 'P1', treat: [{ value: '洁治' }] };
const planned = { id: 'T1', customer_id: 'P1', mode_name: '种植' };
const queries: string[][] = [];
const prisma = {
host: { findFirst: async () => ({ id: 'host', name: 'friday' }) },
patientTransaction: { findMany: async ({ where }: { where: { action: { in: string[] } } }) => {
const actions = where.action?.in ?? [];
queries.push(actions);
return [
...(actions.includes('treatment_completed') ? [{ rawPayload: actual, payloadHash: 'actual' }] : []),
...(actions.includes('treatment_planned') ? [{ rawPayload: planned, payloadHash: 'planned' }] : []),
];
} },
};
// 停在转换入口检查原表,不触发事实写入。
const stop = new Error('captured');
const run = jest.fn(({ tables }: { tables: Record<string, unknown[]> }) => {
expect(tables.med_emr_info).toEqual([actual]);
expect(tables.customer_treat_plan_item).toEqual([planned]);
throw stop;
});
const svc = new ColdImportService(prisma as never, null as never, null as never,
null as never, null as never, { run } as never, null as never);
await expect(svc.reparseFromTransactions({ dir: join(__dirname, '../data/friday'), hostName: 'friday',
subjectTypes: ['treatment'], patientIds: ['p1'] })).rejects.toBe(stop);
expect(run).toHaveBeenCalledTimes(1);
expect(queries).toContainEqual(['treatment_planned']);
expect(queries).toContainEqual(['treatment_completed']);
});
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