feat(sync): 回访摄入带上执行客服 —— 按诊所反推客服名册供主管指派

主管指派工单要选人,而 host **不提供「客服归属诊所」字段**,只能按行为反推
「该诊所近 N 月有过回访记录的客服」。DW fact_returnvisit_out 一直带 task_director_id/name,
PAC 侧此前没映射 —— 花名册只能靠离线快照 data/jvs-dw/users.json(已陈旧、无刷新机制,
之前排查生产操作人时 10 个 id 有 8 个查不到姓名就是这个原因)。

【与「专属客服」是两回事,刻意不合并】
  patients.preferences.dedicatedCs  ← fact_client_out.current_task_director(患者挂谁名下)
  patient_return_visits.task_director_* ← 本次回访是谁做的
两者可以是不同人;口径差 2.7 倍(回访表 5,110 个 distinct 客服 vs 患者表 1,865 个)——
回访是操作留痕,覆盖更全,97.2% 的专属客服在此出现过。合并成一列会让名册少掉大半人。

改动(三层,通用代码零改动):
  - manifest query 补 SELECT 两列(漏了则映射静默失效 → 落 null)
  - assembler 映射 taskDirectorId / taskDirectorName
  - canonical 加两字段(id 用 z.coerce.string:host 是 Int64,PAC 的 external 标识一律字符串)
  - schema + migration:可空两列(166.7 万存量加列不重写表)+ 名册复合索引
    (host, tenant, clinic, task_director, task_date desc) —— 单列不够,clinic 基数仅 64
  - upsert 落库经 emptyToNull:空串 / "0" 归 null,否则名册会多出假的"无名客服"分组

【存量怎么补(本次不含)】reparse **无效** —— 回访是 upsert 资源、不进 transaction,
没有 rawPayload 可重放;整表重摄又会连带重摄这批患者的病历/结算/预约(2026-08-01 实测
那条路把测试服磁盘写满)。存量走一次性 DW 回填:按 external_id 批量 UPDATE 两列。

测试 806 项(+7),锁住"两处客服字段不互相挪用"与源 query 必须选这两列。

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

apps/pac-service/data/jvs-dw/assemblers/patient_return_visit.yaml

@@ -20,3 +20,8 @@ field_mapping:
   treatmentItems:    treatment_items_full   # 大类·子项 合并(transforms H);单字段展示,不镜像 host 两列
   followContent:     follow_content
   result:            return_visit_result
+  # 执行本次回访的客服(≠ 患者主档的 dedicatedCs「专属客服」,可以是不同人)。
+  #   给"按诊所取客服名册 → 主管指派"用:host 无「客服归属诊所」字段,只能按回访行为反推。
+  #   ⚠️ 客服与诊所是多对多(实测 24% 跨诊所),名册里同一人会出现在多个诊所下,这是正常的。
+  taskDirectorId:    task_director_id
+  taskDirectorName:  task_director

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

@@ -288,10 +288,15 @@ sql_source:
     # customer_id AS patient_id(让 cohort 过滤生效);WHERE org ∈ EMR 表出现的所有诊所
     #   (随 DW 全量,当前 64 家,非 5 家)—— 与核心数据同范围。
     # 非召回信号,详情页"回访记录"块展示(常规/术后/咨询回访)。
+    # task_director_id/task_director:**执行本次回访的客服**(≠ 患者主档的"专属客服")。
+    #   用途:按诊所反推客服名册供主管指派 —— host 不提供"客服归属诊所"字段,只能按回访行为反推。
+    #   实测口径:本表 5,110 个 distinct 客服,患者表 current_task_director 只有 1,865 个 —— 回访是
+    #   操作留痕,覆盖更全(97.2% 的专属客服在此出现过)。
     fact_returnvisit_out: |
       SELECT id, customer_id AS patient_id, brand, organization_id, task_date,
              return_visit_type_name, return_visit_status_name, task_status_name,
-             treatment_items, treatment_items_two, follow_content, return_visit_result
+             treatment_items, treatment_items_two, follow_content, return_visit_result,
+             task_director_id, task_director
       FROM dw_group.fact_returnvisit_out
       WHERE organization_id IN (SELECT DISTINCT organization_id FROM dw_group.fact_emr_treatment_out)
 

apps/pac-service/prisma/migrations/20260801150000_return_visit_task_director/migration.sql

+-- 回访执行客服(patient_return_visits.task_director_id / _name)
+--
+-- 【要解决什么】主管指派工单时要选人,但 host **不提供"客服归属诊所"字段** —— 只能按行为反推:
+--   「该诊所近 N 月有过回访记录的客服」。DW fact_returnvisit_out 早就带 task_director_id/name,
+--   PAC 侧此前没映射,于是花名册只能靠离线快照 data/jvs-dw/users.json(已陈旧,无刷新机制)。
+--
+-- 【为什么不是"专属客服"】患者主档的 current_task_director 落在 patients.preferences.dedicatedCs,
+--   语义是"这个患者当前挂在谁名下";本字段是"**这次回访是谁做的**",两者可以是不同人。
+--   实测口径差异也很大:回访表 5,110 个 distinct 客服 vs 患者表 1,865 个(2.7 倍)——
+--   回访是操作留痕,覆盖更全。故各存各的,不复用同一列。
+--
+-- 【索引】(host, tenant, clinic, task_director, task_date desc) 正是名册查询的形状:
+--   按诊所圈人 + 时间窗过滤 + 去重客服。单列索引不够(clinic 基数只有 64,选择性差)。
+--
+-- 【存量】加可空列不重写表(PG11+ 元数据操作),存量行为 NULL —— 名册查询自动跳过它们。
+--   166.7 万存量行走一次性 DW 回填(按 external_id 批量 UPDATE),不走 cold-import:
+--   回访是 upsert 资源、**不进 transaction**,所以 reparse 无效;而整表重摄会连带
+--   重摄这批患者的病历/结算/预约(2026-08-01 实测:那条路把测试服磁盘写满了)。
+ALTER TABLE "patient_return_visits"
+  ADD COLUMN IF NOT EXISTS "task_director_id"   TEXT,
+  ADD COLUMN IF NOT EXISTS "task_director_name" TEXT;
+
+CREATE INDEX IF NOT EXISTS "patient_return_visits_roster_idx"
+  ON "patient_return_visits" ("host_id", "tenant_id", "clinic_id", "task_director_id", "task_date" DESC);

apps/pac-service/prisma/schema.prisma

@@ -417,6 +417,14 @@ model PatientReturnVisit {
   /// 回访结果
   result         String?
 
+  /// 执行本次回访的客服(host 侧 task_director)—— **不是**患者的"专属客服"
+  /// (那个在 patients.preferences.dedicatedCs,可以是另一个人)。
+  /// 用途:按诊所反推客服名册(该诊所近 N 月有回访行为的人)供主管指派 —— host 不提供
+  /// "客服归属诊所"字段,只能按行为反推。⚠️ 客服与诊所是**多对多**(实测 24% 跨诊所,
+  /// 最多跨 13 家),同一人会出现在多个诊所名册里,不要当成一对一属性建模。
+  taskDirectorId   String? @map("task_director_id")
+  taskDirectorName String? @map("task_director_name")
+
   createdAt DateTime @default(now()) @map("created_at") @db.Timestamptz(3)
   updatedAt DateTime @updatedAt @map("updated_at") @db.Timestamptz(3)
 
@@ -424,6 +432,8 @@ model PatientReturnVisit {
 
   @@unique([hostId, tenantId, sourceUnit, externalId])
   @@index([patientId, taskDate(sort: Desc)])
+  /// 客服名册查询:按 (诊所, 客服) 聚合 + taskDate 卡时间窗
+  @@index([hostId, tenantId, clinicId, taskDirectorId, taskDate(sort: Desc)])
   @@map("patient_return_visits")
 }
 

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

@@ -1795,6 +1795,11 @@ export class ColdImportService {
         treatmentItems: normalizeMergedItems(c.treatmentItems as string | undefined),
         followContent: (c.followContent as string | undefined) ?? null,
         result: (c.result as string | undefined) ?? null,
+        // 执行本次回访的客服(≠ 患者的"专属客服")。给"按诊所取客服名册 → 主管指派"用。
+        //   host 侧 id 是数字,canonical 已 coerce 成 string(同 externalId 口径:external 标识一律字符串)。
+        //   空串归一为 null —— 名册查询按 IS NOT NULL 圈人,留空串会混进一个"无名客服"分组。
+        taskDirectorId: emptyToNull(c.taskDirectorId),
+        taskDirectorName: emptyToNull(c.taskDirectorName),
       };
       try {
         await this.withDbRetry(
@@ -2618,6 +2623,15 @@ function shiftIsoHours(value: string, deltaHours: number): string {
   return new Date(d.getTime() + deltaHours * 3600_000).toISOString();
 }
 
+/// 空值归一:undefined / null / 空白串 / host 的数字 0 → null;其余 trim 后返回字符串。
+/// 用于 external 标识类字段(如回访客服 id)—— 名册查询按 IS NOT NULL 圈人,
+/// 留下空串或 "0" 会各自成为一个假的"客服"分组。
+function emptyToNull(v: unknown): string | null {
+  if (v === undefined || v === null) return null;
+  const s = String(v).trim();
+  return s === '' || s === '0' ? null : s;
+}
+
 /// 回访治疗项「大类 · 子项」合并值归一(transforms H 的 concat)。
 /// 两列皆空 → 合并出来只剩 " · "(纯分隔符/空白)→ 视为无治疗项,归 null;有真实内容则 trim 返回。
 function normalizeMergedItems(v: string | undefined): string | null {

apps/pac-service/tests/return-visit-task-director.spec.ts

+import { readFileSync } from 'node:fs';
+import { join } from 'node:path';
+import * as yaml from 'js-yaml';
+import { PatientReturnVisitCanonicalSchema } from '@pac/types';
+
+/**
+ * 回访执行客服(patient_return_visits.task_director_id / _name)契约回归。
+ *
+ * 【为什么要这个字段】主管指派工单要选人,而 host **不提供「客服归属诊所」字段** ——
+ * 只能按行为反推「该诊所近 N 月有过回访记录的客服」。DW fact_returnvisit_out 一直带
+ * task_director_id/name,PAC 侧此前没映射,花名册只能靠离线快照 data/jvs-dw/users.json
+ * (已陈旧且无刷新机制)。
+ *
+ * 【最容易搞错的一点】它**不是**患者的"专属客服":
+ *   - patients.preferences.dedicatedCs ← fact_client_out.current_task_director(患者挂在谁名下)
+ *   - patient_return_visits.task_director_* ← 本次回访是谁做的
+ *   两者可以是不同人,口径也差 2.7 倍(回访表 5,110 个 distinct vs 患者表 1,865 个)。
+ *   谁要是把两处合并成一列,名册就会少掉一大半人 —— 故这组测试显式锁住"各存各的"。
+ */
+
+const RV_YAML = join(__dirname, '../data/jvs-dw/assemblers/patient_return_visit.yaml');
+const MANIFEST = join(__dirname, '../data/jvs-dw/manifest.yaml');
+
+describe('canonical schema', () => {
+  test('⭐ host 侧数字 id → 字符串(PAC 的 external 标识一律字符串,同 externalId 口径)', () => {
+    const out = PatientReturnVisitCanonicalSchema.parse({
+      externalId: 'rv1',
+      patientExternalId: 'p1',
+      taskDirectorId: 5679, // DW 是 Int64
+      taskDirectorName: '李欣',
+    });
+    expect(out.taskDirectorId).toBe('5679');
+    expect(out.taskDirectorName).toBe('李欣');
+  });
+
+  test('未提供 → undefined(不是空串);不因缺这两列而拒收整条回访', () => {
+    const out = PatientReturnVisitCanonicalSchema.parse({
+      externalId: 'rv2',
+      patientExternalId: 'p2',
+    });
+    expect(out.taskDirectorId).toBeUndefined();
+    expect(out.externalId).toBe('rv2'); // 主数据完好
+  });
+});
+
+describe('yaml 契约', () => {
+  const rv = yaml.load(readFileSync(RV_YAML, 'utf-8')) as {
+    canonical: string;
+    field_mapping: Record<string, string>;
+  };
+
+  test('assembler 映射到 host 的 task_director_id / task_director', () => {
+    expect(rv.canonical).toBe('patient_return_visit');
+    expect(rv.field_mapping.taskDirectorId).toBe('task_director_id');
+    expect(rv.field_mapping.taskDirectorName).toBe('task_director');
+  });
+
+  test('⭐ 源 query 必须 SELECT 这两列 —— 漏了则映射静默失效(canonical 无此键 → 落 null)', () => {
+    const manifest = yaml.load(readFileSync(MANIFEST, 'utf-8')) as {
+      sql_source?: { queries?: Record<string, string> };
+    };
+    const sql = manifest.sql_source?.queries?.['fact_returnvisit_out'] ?? '';
+    expect(sql).toMatch(/\btask_director_id\b/);
+    expect(sql).toMatch(/\btask_director\b/);
+  });
+
+  test('⭐ 患者主档的「专属客服」仍走自己的字段,没被本次改动挪用', () => {
+    const p = yaml.load(
+      readFileSync(join(__dirname, '../data/jvs-dw/assemblers/patient.yaml'), 'utf-8'),
+    ) as { field_mapping: Record<string, string> };
+    expect(p.field_mapping.dedicatedCsName).toBe('current_task_director');
+    expect(p.field_mapping.dedicatedCsId).toBe('current_task_director_id');
+    // 两处映射的 host 列不同 —— 合并会让名册少掉 2/3 的人
+    expect(p.field_mapping.dedicatedCsId).not.toBe(rv.field_mapping.taskDirectorId);
+  });
+});
+
+describe('schema.prisma', () => {
+  const schema = readFileSync(join(__dirname, '../prisma/schema.prisma'), 'utf-8');
+  const model = schema.slice(
+    schema.indexOf('model PatientReturnVisit'),
+    schema.indexOf('@@map("patient_return_visits")'),
+  );
+
+  test('两列都是可空(存量 166 万行加列不重写表,回填前保持 NULL)', () => {
+    expect(model).toMatch(/taskDirectorId\s+String\?\s+@map\("task_director_id"\)/);
+    expect(model).toMatch(/taskDirectorName\s+String\?\s+@map\("task_director_name"\)/);
+  });
+
+  test('⭐ 名册查询的复合索引存在 —— 单列索引不够(clinic 基数仅 64,选择性差)', () => {
+    expect(model).toMatch(/@@index\(\[hostId, tenantId, clinicId, taskDirectorId, taskDate/);
+  });
+});

packages/types/src/schemas/canonical.ts

@@ -114,6 +114,11 @@ export const PatientReturnVisitCanonicalSchema = z
     treatmentItems: z.string().optional().nullable(),
     followContent: z.string().optional().nullable(),
     result: z.string().optional().nullable(),
+    /// 执行本次回访的客服 —— **不是**患者的"专属客服"(那个是 patient.dedicatedCs*,可以是另一个人)。
+    /// 用途:按诊所反推客服名册供主管指派(host 无"客服归属诊所"字段,只能按回访行为反推)。
+    /// id 用 string:host 侧是数字,但 PAC 一律按字符串存 external 标识(同 externalId 口径)。
+    taskDirectorId: z.coerce.string().optional().nullable(),
+    taskDirectorName: z.string().optional().nullable(),
   })
   .passthrough();
 export type PatientReturnVisitCanonical = z.infer<typeof PatientReturnVisitCanonicalSchema>;