feat(sync): 新增 lookup transform + FRIDAY patient 首版接入

lookup operator(Layer A.5 第 8 个):按外键 join 子表、按 order_by 挑一条、
把字段附加到主表——补上"1:N 子表归约成标量"的空(join_arrays 只产数组,
其余 operator 全是单表内变换)。纯增量,不改任何现有 operator,不影响 jvs-dw。

FRIDAY(第二宿主)patient 首版:方案 B(tenant=friday-market + 品牌 source_unit),
宿主纯导原始表 customer_basic_info + customer_contacts,phone 由 lookup 挑默认号
(is_default→本人→最早)填入。本地 450 患者/2 品牌真实落库验证:source_unit、
挑号(is_default 胜出)、gender/病历号/时区全对。CSV 含 PII 不入 git。

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

apps/pac-service/data/friday/.gitignore

+# FRIDAY 原始导出数据含真实 PII(姓名/电话/病历)—— 一律不入 git。
+# 只提交 PAC 侧的映射配置(manifest.yaml + assemblers/*.yaml)。
+*
+!.gitignore
+!manifest.yaml
+!assemblers
+!assemblers/*.yaml

apps/pac-service/data/friday/assemblers/patient.yaml

+# patient — 主档 upsert(不进 transaction,无 emits)
+# 源:customer.customer_basic_info(原始表);phone 由 manifest.transforms.lookup
+#     从 customer_contacts 挑默认号附加进来,故这里直接映射标量 phone。
+canonical: patient
+
+primary:
+  table: customer_basic_info
+  key: id
+
+field_mapping:
+  externalId:          id             # customer_basic_info 主键 = host 内部患者唯一 id(→ String 化)
+  name:                name
+  phone:               phone          # 由 transforms.lookup 从 customer_contacts 挑默认号附加
+  gender:              sex            # tinyint 1/2 → enum_mapping 归一;0/空 → 空
+  birthDate:           birthday       # date;空值入库为 null
+  medicalRecordNumber: file_number    # 档案号(病历号,客服沟通用)
+  createdAt:           created_gmt_at
+  updatedAt:           updated_gmt_at
+  # source_unit(品牌)由 manifest.identity_namespace_field=brand_id 填,不在此声明。
+  # organization_id / status / 获客渠道 等首版不接,后续资源(encounter/settlement)接入再补。
+
+# 性别:host tinyint → PAC gender(自由文本,前端 formatGender 认 male/female)
+enum_mapping:
+  gender:
+    '1':      male
+    '2':      female
+    _default: ''      # 0 / 空 / 未知 → 空(gender 可空,不阻挡入库)

apps/pac-service/data/friday/manifest.yaml

+# FRIDAY SaaS Cold Import Manifest(首版 — 仅 patient 试跑)
+#
+# 数据源:FRIDAY SaaS MySQL(customer 库)。cold-import 不支持 MySQL 直连,
+#   走 CSV 文件模式:host 端**纯导原始表**(SELECT * FROM tb),不做任何 join。
+#   电话在 1:N 的 customer_contacts 里 → 由 transforms.lookup 按 customer_id join、
+#   挑默认号(is_default→本人→最早)填成 patient.phone 标量。宿主零特殊处理。
+#
+# ── 方案 B:合成"市场"集团 tenant;品牌降为 source_unit 命名空间 ──
+#   FRIDAY 是多品牌 SaaS(30+ 独立品牌 tenant_id,各含 1-8 诊所 organization_id)。
+#   PAC tenant = 合成 friday-market(整个市场一个租户,便于 FRIDAY 作为运营方跨品牌召回);
+#   品牌(源 brand_id 列)→ patients.source_unit(集团内给 external_id 消歧:同号不同品牌=不同人);
+#   诊所 organization_id → 后续 orgScope 隔离维度。
+#   ⚠️ 隔离靠 orgScope/source_unit,非 tenant 边界 → 生产必须「禁止空 orgScope」。
+#      若要品牌级硬隔离,改为 tenant_field: brand_id(每品牌一个 PAC tenant)。
+#
+# 跑法:
+#   pnpm sync -- --dir=./data/friday --dry-run     # 看翻译效果,不写库
+#   pnpm sync -- --dir=./data/friday --no-recompute
+
+host_name:    friday
+display_name: FRIDAY SaaS(多品牌市场)
+auto_sync:    false
+
+tenant_id:                friday-market   # PAC 租户 = 合成市场根(静态)
+identity_namespace_field: brand_id        # patients.source_unit ← 源品牌列(customer.tenant_id 导出别名)
+
+amount_unit: yuan
+timezone:    Asia/Shanghai
+
+tables:
+  - { table: customer_basic_info, file: customer_basic_info.csv }   # 患者主档(原始)
+  - { table: customer_contacts,   file: customer_contacts.csv }     # 联系方式 1:N(原始)
+
+# Layer A.5:把 1:N 联系方式归约成 patient.phone 标量(无需源侧 join)
+transforms:
+  - kind: lookup
+    input: customer_basic_info
+    output: customer_basic_info        # in-place 附加 phone 列
+    from: customer_contacts
+    left_key: id                       # 患者主键
+    right_key: customer_id             # 联系方式指向患者的外键
+    where:
+      contacts_tel: { not_empty: true } # 空号候选剔除
+    order_by:                          # 挑一条:默认号 → 本人(type=1) → 最早 id
+      - { field: is_default,    dir: desc }
+      - { field: contacts_type, dir: asc }
+      - { field: id,            dir: asc }
+    select:
+      phone: contacts_tel              # patient.phone ← 选中候选的 contacts_tel
+
+assemblers:
+  - { file: assemblers/patient.yaml }

apps/pac-service/src/modules/sync/transforms/operators/lookup.op.ts

+import type { LookupOp, LookupWhereRule } from '../transforms.schema';
+import { SOURCE_ROW_KEY, isEmptyText, type Row } from '../row';
+
+/**
+ * lookup — 按外键 join `from` 子表,挑一条,把 select 字段附加到主表(1:N → 标量)。
+ *
+ * 左连接语义:输出行数 = input 行数;未命中的行 select 目标字段填 null(保证列存在,
+ * 下游 field_mapping 稳定)。多条候选按 order_by 稳定排序取第一条;不配则取首个遇到。
+ *
+ * __source_row 透传主表自身(不合并 from 的),合规留底不变。
+ */
+export function runLookup(
+  op: LookupOp,
+  input: Row[],
+  from: Row[],
+): { rows: Row[]; matched: number; missed: number } {
+  // 1. 建 from 索引(right_key → 候选行);建索引时先跑 where 预过滤
+  const index = new Map<string, Row[]>();
+  for (const r of from) {
+    if (op.where && !passLookupWhere(r, op.where)) continue;
+    const k = keyOf(r[op.right_key]);
+    if (k === null) continue;
+    const bucket = index.get(k);
+    if (bucket) bucket.push(r);
+    else index.set(k, [r]);
+  }
+
+  // 2. 逐主表行取候选、挑一条、附加字段
+  let matched = 0;
+  let missed = 0;
+  const rows = input.map((row) => {
+    const out: Row = { ...row };
+    const k = keyOf(row[op.left_key]);
+    const candidates = k === null ? undefined : index.get(k);
+    if (candidates && candidates.length > 0) {
+      const chosen = pickOne(candidates, op.order_by);
+      for (const [target, fromField] of Object.entries(op.select)) {
+        out[target] = chosen[fromField] ?? null;
+      }
+      matched++;
+    } else {
+      // 未命中:填 null(仅当主表尚无该列,不覆盖已有值)
+      for (const target of Object.keys(op.select)) {
+        if (!(target in out)) out[target] = null;
+      }
+      missed++;
+    }
+    // __source_row 保留主表自身(spread 已带,显式兜底一次防被 select 覆盖同名 key)
+    if (SOURCE_ROW_KEY in row) out[SOURCE_ROW_KEY] = row[SOURCE_ROW_KEY];
+    return out;
+  });
+
+  return { rows, matched, missed };
+}
+
+/// join 键归一:null/undefined/空串 → null(不参与匹配);其余 String 化对齐(CSV 全是字符串)
+function keyOf(v: unknown): string | null {
+  if (v === undefined || v === null) return null;
+  const s = String(v).trim();
+  return s === '' ? null : s;
+}
+
+/// 候选排序取第一条(order_by 为空 → 首个遇到)。稳定:多键顺序裁决,数值优先按数值比。
+function pickOne(cands: Row[], orderBy?: LookupOp['order_by']): Row {
+  if (!orderBy || orderBy.length === 0) return cands[0]!;
+  const sorted = [...cands].sort((a, b) => {
+    for (const { field, dir } of orderBy) {
+      const cmp = compareVals(a[field], b[field]);
+      if (cmp !== 0) return dir === 'desc' ? -cmp : cmp;
+    }
+    return 0;
+  });
+  return sorted[0]!;
+}
+
+function toNum(v: unknown): number | null {
+  if (typeof v === 'number') return Number.isFinite(v) ? v : null;
+  if (typeof v === 'string' && v.trim() !== '' && !Number.isNaN(Number(v))) return Number(v);
+  return null;
+}
+
+/// 比较:两侧都能当数字 → 数值比;否则字符串比。null/空当最小(排后需配 dir)。
+function compareVals(a: unknown, b: unknown): number {
+  const na = toNum(a);
+  const nb = toNum(b);
+  if (na !== null && nb !== null) return na - nb;
+  const sa = a === null || a === undefined ? '' : String(a);
+  const sb = b === null || b === undefined ? '' : String(b);
+  return sa < sb ? -1 : sa > sb ? 1 : 0;
+}
+
+/// 候选行预过滤(AND;标量子集:not_empty / empty / in / equals)
+function passLookupWhere(row: Row, where: Record<string, LookupWhereRule>): boolean {
+  for (const [field, rule] of Object.entries(where)) {
+    const v = row[field];
+    if ('not_empty' in rule) {
+      if (isEmptyText(v)) return false;
+    } else if ('empty' in rule) {
+      if (!isEmptyText(v)) return false;
+    } else if ('in' in rule) {
+      const set = new Set(rule.in.map((x) => String(x)));
+      if (!set.has(String(v))) return false;
+    } else if ('equals' in rule) {
+      if (String(v) !== String(rule.equals)) return false;
+    }
+  }
+  return true;
+}

apps/pac-service/src/modules/sync/transforms/transform-engine.ts

@@ -5,6 +5,7 @@ import { runDerive } from './operators/derive.op';
 import { runRouteByPattern } from './operators/route-by-pattern.op';
 import { runPickFirstNonzero } from './operators/pick-first-nonzero.op';
 import { runFilter } from './operators/filter.op';
+import { runLookup } from './operators/lookup.op';
 import { SOURCE_ROW_KEY, type TableMap } from './row';
 import type { TransformsConfig } from './transforms.schema';
 
@@ -119,6 +120,21 @@ export class TransformEngine {
           );
           break;
         }
+        case 'lookup': {
+          const from = tables[op.from];
+          if (!from) {
+            this.logger.warn(
+              `[transform/lookup] from 表 "${op.from}" 不存在;当空表(全部未命中,填 null)`,
+            );
+          }
+          const { rows, matched, missed } = runLookup(op, input, from ?? []);
+          tables[op.output] = rows;
+          this.logger.log(
+            `[transform/lookup] "${inputName}"(${input.length}) ←(${op.from} ${from?.length ?? 0}) ` +
+              `→ "${op.output}" select=${Object.keys(op.select).join(',')} matched=${matched} missed=${missed}`,
+          );
+          break;
+        }
       }
     }
 

apps/pac-service/src/modules/sync/transforms/transforms.schema.ts

@@ -302,7 +302,60 @@ export const UnionOpSchema = z.object({
 export type UnionOp = z.infer<typeof UnionOpSchema>;
 
 // ─────────────────────────────────────────────────────────
-// 7 operator union + transforms[] 顶层
+// ⑧ lookup — 按外键 join 子表,挑一条,把字段附加到主表(1:N → 标量)
+// ─────────────────────────────────────────────────────────
+
+/// lookup 的候选行预过滤规则(标量子集,不含 filter 的 JSON 数组规则)。
+const LookupWhereRuleSchema = z.union([
+  z.object({ not_empty: z.literal(true) }),
+  z.object({ empty: z.literal(true) }),
+  z.object({ in: z.array(z.union([z.string(), z.number()])).min(1) }),
+  z.object({ equals: z.union([z.string(), z.number()]) }),
+]);
+export type LookupWhereRule = z.infer<typeof LookupWhereRuleSchema>;
+
+/**
+ * 把 `from` 子表按外键 join 到 `input` 主表,多条候选时按 order_by 挑第一条,
+ * 把 select 里的字段附加到主表行(等价 LEFT JOIN + 取一条 + 选列)。
+ *
+ * 用途:1:N 子表归约成主表**标量**字段 —— join_arrays 只能产数组、transforms 其余
+ *   operator 全是单表内变换,都填不了这个空。典型:患者主档(customer_basic_info)从
+ *   联系方式子表(customer_contacts)取"默认电话"填 patient.phone。
+ *
+ * 语义:
+ *   - 输出行数 = input 行数(左连接:未命中 → select 目标字段填 null,保证列存在)
+ *   - order_by 多键稳定排序,第一条胜出;不配 → 取该外键下"首个遇到"的候选
+ *     例:[{field: is_default, dir: desc}, {field: contacts_type, dir: asc}, {field: id, dir: asc}]
+ *         → 优先默认号,再优先本人(type=1<2),再取最早 id
+ *   - where(可选)对候选行预过滤(AND;如 contacts_tel not_empty)
+ *   - select:`目标字段(主表列名) ← 来源字段(from 列名)`,与 field_mapping 同向
+ *   - __source_row 保留主表的(不合并 from 的),合规留底不变
+ */
+export const LookupOpSchema = z.object({
+  kind: z.literal('lookup'),
+  input: tableNameSchema,
+  output: tableNameSchema,
+  from: tableNameSchema.describe('被 join 的子表(需在 tables 或前置 transform 产出里)'),
+  left_key: z.string().min(1).describe('input 行的关联键字段'),
+  right_key: z.string().min(1).describe('from 行里指向 input 的外键字段'),
+  where: z.record(z.string(), LookupWhereRuleSchema).optional().describe('对候选行预过滤(AND)'),
+  order_by: z
+    .array(
+      z.object({
+        field: z.string().min(1),
+        dir: z.enum(['asc', 'desc']).default('asc'),
+      }),
+    )
+    .optional()
+    .describe('多候选排序,第一条胜出;不配 = 首个遇到'),
+  select: z
+    .record(z.string(), z.string())
+    .describe('附加字段:目标字段(主表列名) ← 来源字段(from 列名)'),
+});
+export type LookupOp = z.infer<typeof LookupOpSchema>;
+
+// ─────────────────────────────────────────────────────────
+// operator union + transforms[] 顶层
 // ─────────────────────────────────────────────────────────
 
 export const TransformOpSchema = z.discriminatedUnion('kind', [
@@ -313,6 +366,7 @@ export const TransformOpSchema = z.discriminatedUnion('kind', [
   PickFirstNonzeroOpSchema,
   FilterOpSchema,
   UnionOpSchema,
+  LookupOpSchema,
 ]);
 export type TransformOp = z.infer<typeof TransformOpSchema>;
 

apps/pac-service/tests/transforms.spec.ts

@@ -13,6 +13,7 @@ import { runDerive } from '../src/modules/sync/transforms/operators/derive.op';
 import { runRouteByPattern } from '../src/modules/sync/transforms/operators/route-by-pattern.op';
 import { runPickFirstNonzero } from '../src/modules/sync/transforms/operators/pick-first-nonzero.op';
 import { runFilter } from '../src/modules/sync/transforms/operators/filter.op';
+import { runLookup } from '../src/modules/sync/transforms/operators/lookup.op';
 
 describe('transforms | project', () => {
   test('keep + rename 选列改名', () => {
@@ -393,6 +394,74 @@ describe('transforms | filter', () => {
 // TransformEngine 集成
 // ─────────────────────────────────────────────
 
+describe('transforms | lookup', () => {
+  const op = {
+    kind: 'lookup' as const,
+    input: 'patient',
+    output: 'patient',
+    from: 'contacts',
+    left_key: 'id',
+    right_key: 'customer_id',
+    where: { contacts_tel: { not_empty: true as const } },
+    order_by: [
+      { field: 'is_default', dir: 'desc' as const },
+      { field: 'contacts_type', dir: 'asc' as const },
+      { field: 'id', dir: 'asc' as const },
+    ],
+    select: { phone: 'contacts_tel' },
+  };
+
+  test('多候选按 order_by 挑默认号(is_default desc)', () => {
+    const { rows, matched, missed } = runLookup(op, [{ id: '1', name: 'A' }], [
+      { id: '10', customer_id: '1', contacts_tel: '111', is_default: '0', contacts_type: '1' },
+      { id: '11', customer_id: '1', contacts_tel: '222', is_default: '1', contacts_type: '2' },
+    ]);
+    expect(rows[0]).toEqual({ id: '1', name: 'A', phone: '222' }); // is_default=1 胜出
+    expect(matched).toBe(1);
+    expect(missed).toBe(0);
+  });
+
+  test('平手时按 contacts_type asc(本人=1 优先),再 id asc', () => {
+    const { rows } = runLookup(op, [{ id: '1' }], [
+      { id: '20', customer_id: '1', contacts_tel: '联系人', is_default: '0', contacts_type: '2' },
+      { id: '21', customer_id: '1', contacts_tel: '本人', is_default: '0', contacts_type: '1' },
+    ]);
+    expect(rows[0]!.phone).toBe('本人');
+  });
+
+  test('未命中 → phone 填 null(左连接,行数不变)', () => {
+    const { rows, matched, missed } = runLookup(op, [{ id: '99', name: 'Z' }], [
+      { id: '10', customer_id: '1', contacts_tel: '111', is_default: '1', contacts_type: '1' },
+    ]);
+    expect(rows).toEqual([{ id: '99', name: 'Z', phone: null }]);
+    expect(matched).toBe(0);
+    expect(missed).toBe(1);
+  });
+
+  test('where 预过滤:空号码候选被排除', () => {
+    const { rows } = runLookup(op, [{ id: '1' }], [
+      { id: '10', customer_id: '1', contacts_tel: '', is_default: '1', contacts_type: '1' }, // 空号被 where 剔除
+      { id: '11', customer_id: '1', contacts_tel: '333', is_default: '0', contacts_type: '1' },
+    ]);
+    expect(rows[0]!.phone).toBe('333');
+  });
+
+  test('输入不可变(不污染上游 input 行)', () => {
+    const input = [{ id: '1', name: 'A' }];
+    runLookup(op, input, [{ id: '10', customer_id: '1', contacts_tel: '111', is_default: '1', contacts_type: '1' }]);
+    expect(input[0]).toEqual({ id: '1', name: 'A' }); // 原行没被加 phone
+  });
+
+  test('引擎集成:from 表缺失 → 全部 null,不抛错', () => {
+    const engine = new TransformEngine();
+    const result = engine.run({
+      tables: { patient: [{ id: '1' }] },
+      transforms: [{ ...op, where: undefined, order_by: undefined }],
+    });
+    expect(result.patient![0]).toMatchObject({ id: '1', phone: null });
+  });
+});
+
 describe('transforms | __source_row 真原文留底', () => {
   test('TransformEngine 入口给每行打 __source_row=自身', () => {
     const engine = new TransformEngine();