feat(sync): cold-import 新增 --clinics=X,Y 按诊所收窄 cohort(不动 manifest SQL)

需求:数仓加了新诊所数据,要挑几家追加摄入,不清空重摄、且摄该诊所患者的全部相关主体。
- manifest schema:cohort 新增可选 clinic_scope {from_table, org_column}(被动声明)。
- cold-import CLI:--clinics=X,Y;service 透传;listPatientPairs 注入诊所过滤。
  只收窄"列哪些患者"(挂在有 org 的事实表上求交集),不碰任何资源查询 →
  被选中患者的全部主体(跨所有诊所)照常摄入,幂等追加不清空、不误删。
- jvs-dw manifest:一次性配 clinic_scope→fact_emr_treatment_out.organization_id。
- 默认(不传 --clinics)全量行为完全不变。DW 实测:单家 27194 / 两家 60992 vs 全量 569 万,收窄正确。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

apps/pac-service/data/jvs-dw/manifest.yaml

@@ -103,6 +103,12 @@ sql_source:
     patient_key_column:  patient_id                # 患者主键列(所有源表共用此列名做 cohort 过滤)
     tenant_key_column:   brand                     # 租户区分列(jvs-dw 多 brand;单 tenant host 删本行)
     list_cursor_column:  last_visit_time           # 列患者时增量 cursor 列(对应患者主档表时间列)
+    # 一次性声明:cold-import `--clinics=X,Y` 据此把 cohort 收窄到"在这些诊所看过"的患者。
+    # 患者主档 fact_client_out 无 org → 诊所过滤挂在有 org 的治疗事实表上。被动配置:
+    # 不传 --clinics 时完全不生效(默认全量);只收窄"列哪些患者",资源查询不动(不误删)。
+    clinic_scope:
+      from_table:  dw_group.fact_emr_treatment_out
+      org_column:  organization_id
 
   # SQL 最朴素化 — host(DW)给的数据范围就是 PAC 该消化的范围。
   # PAC 这边不预设诊所 / brand / 时间过滤,数据来什么就是什么。

apps/pac-service/src/cli/cold-import.cli.ts

@@ -22,6 +22,7 @@ interface CliArgs {
   help: boolean;
   incremental: boolean;
   cohortBatchSize?: number | null; // null = 显式禁用分批,undefined = 用 env / 默认
+  clinics?: string[]; // --clinics=X,Y:只列"在这些诊所看过"的患者(其全部相关主体仍全摄);不传=全量
 }
 
 function parseArgs(argv: string[]): CliArgs {
@@ -34,6 +35,12 @@ function parseArgs(argv: string[]): CliArgs {
     else if (a.startsWith('--cohort-batch=')) {
       const n = parseInt(a.slice('--cohort-batch='.length), 10);
       args.cohortBatchSize = Number.isFinite(n) && n >= 0 ? n : undefined;
+    } else if (a.startsWith('--clinics=')) {
+      args.clinics = a
+        .slice('--clinics='.length)
+        .split(',')
+        .map((s) => s.trim())
+        .filter(Boolean);
     } else if (a.startsWith('--dir=')) args.dir = a.slice('--dir='.length);
   }
   return args;
@@ -55,6 +62,8 @@ function printHelp() {
     '  --incremental         增量模式(读上次 cursor,SQL 注 WHERE > cursor;首跑等价全量)',
     '  --cohort-batch=<N>    按 patient 分批跑(default 5000;env PAC_COHORT_BATCH_SIZE 兜底)',
     '  --no-cohort           显式禁用分批,跑 single-shot(文件源或调试场景)',
+    '  --clinics=<id,id>     只列"在这些诊所看过"的患者(其全部相关主体仍全摄;需 manifest',
+    '                        cohort.clinic_scope 配置)。不传=全量。用于按诊所分批补摄,幂等追加不清空。',
     '  --help, -h            显示本帮助',
     '',
     'Examples:',
@@ -91,6 +100,7 @@ async function bootstrap() {
       dryRun: args.dryRun,
       incremental: args.incremental,
       cohortBatchSize: args.cohortBatchSize,
+      clinics: args.clinics,
     });
 
     logger.log('─────────────────────────────────────────');

apps/pac-service/src/modules/sync/cold-import/clickhouse-source.service.ts

@@ -331,6 +331,7 @@ export class ClickHouseSourceService {
   async listPatientPairs(
     source: ClickHouseSource,
     incremental?: IncrementalConfig,
+    clinics?: string[],
   ): Promise<CohortKey[]> {
     const cohort = source.cohort;
     if (!cohort) {
@@ -390,6 +391,28 @@ export class ClickHouseSourceService {
         const quoted = ids.map((id) => `'${id.replace(/'/g, "''")}'`).join(', ');
         whereParts.push(ids.length === 1 ? `${patient_key_column} = ${quoted}` : `${patient_key_column} IN (${quoted})`);
       }
+      // --clinics=X,Y:把 cohort 收窄到「在这些诊所看过」的患者。诊所过滤挂在有 org 的事实表
+      //   (cohort.clinic_scope),用与 cohort 相同的患者键求交集。⚠️ 只收窄「列哪些患者」,
+      //   不碰任何资源查询 → 被选中患者的全部相关主体(跨所有诊所)照常摄入(不误删)。
+      if (clinics && clinics.length > 0) {
+        const cs = cohort.clinic_scope;
+        if (!cs) {
+          throw new Error(
+            '--clinics 需要 manifest.sql_source.cohort.clinic_scope 配置(from_table / org_column)',
+          );
+        }
+        const orgList = clinics.map((c) => `'${c.replace(/'/g, "''")}'`).join(', ');
+        const keyCols = tenant_key_column
+          ? `${patient_key_column}, ${tenant_key_column}`
+          : patient_key_column;
+        const inSubquery = `SELECT DISTINCT ${keyCols} FROM ${cs.from_table} WHERE ${cs.org_column} IN (${orgList})`;
+        whereParts.push(
+          tenant_key_column
+            ? `(${keyCols}) IN (${inSubquery})`
+            : `${patient_key_column} IN (${inSubquery})`,
+        );
+        this.logger.log(`[clickhouse·cohort] --clinics 收窄:${clinics.length} 家诊所 (${cs.from_table}.${cs.org_column})`);
+      }
       const whereSql = whereParts.length > 0 ? ` WHERE ${whereParts.join(' AND ')}` : '';
       // dev/ops:PAC_COHORT_LIMIT=N → 只取 N 个患者(本地抽样重摄用)。不设 = 全量,默认行为不变。
       //   PAC_COHORT_SAMPLE = recent(默认,最近就诊优先)| oldest(最久未来,lapsed,召回候选多)

apps/pac-service/src/modules/sync/cold-import/cold-import.service.ts

@@ -584,6 +584,9 @@ export class ColdImportService {
       /// 按 patient 分批跑(内存友好)。 null/undefined/0 → 单 shot(向后兼容);
       /// 实际生效大小见 resolveCohortBatchSize(env PAC_COHORT_BATCH_SIZE 兜底,默认 5000)
       cohortBatchSize?: number | null;
+      /// --clinics=X,Y:只列"在这些诊所看过"的患者(收窄 cohort;其全部相关主体仍全摄)。
+      /// 需 manifest cohort.clinic_scope 配置。不传/空 → 全量(默认行为不变)。
+      clinics?: string[];
     } = {},
   ): Promise<ImportRunResult> {
     const absDir = path.resolve(dir);
@@ -761,6 +764,7 @@ export class ColdImportService {
         const allPairs = await this.chSource.listPatientPairs(
           sqlSource,
           incrementalConfig,
+          options.clinics,
         );
         if (allPairs.length === 0) {
           // A:增量模式 + cohort 空 → DW 本窗口(已含回看)无任何数据 → 标记不推进游标

apps/pac-service/src/modules/sync/cold-import/manifest.schema.ts

@@ -97,6 +97,18 @@ export const ClickHouseSourceSchema = z.object({
       /// 列患者时的增量 cursor 列(对应 patient_list_from 表的时间列)
       /// 例:last_visit_time;不配 → 列患者不做增量过滤(总是全量列再 cohort 切)
       list_cursor_column: z.string().min(1).optional(),
+      /// 可选:诊所范围过滤源 —— 供 cold-import 的 `--clinics=X,Y` 把 cohort 收窄到
+      ///   「在这些诊所看过的患者」。patient 主档(patient_list_from)通常无 org 列,
+      ///   故诊所过滤挂在有 org 的事实表上。例:
+      ///     { from_table: dw_group.fact_emr_treatment_out, org_column: organization_id }
+      ///   ⚠️ 被动声明:不传 `--clinics` 时完全不生效(默认仍全量)。只收窄「列哪些患者」,
+      ///      不碰任何资源查询 → 被选中患者的全部相关主体(跨所有诊所)照常摄入。
+      clinic_scope: z
+        .object({
+          from_table: z.string().min(1),
+          org_column: z.string().min(1).default('organization_id'),
+        })
+        .optional(),
     })
     .optional(),
 });