merge: 诊所名取现名(argMax by time)→ main

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

@@ -41,6 +41,8 @@ clinic_directory:
   table:      med_emr_info
   id_field:   organization_id
   name_field: organization_name
+  # 按时间取最新(同 jvs-dw 的理由:诊所改名后旧名仍留在历史行里)。
+  time_field: updated_gmt_at
 
 
 # ── 增量水位声明(2026-07 与 jvs-dw 统一;file 源放 manifest 顶层)──

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

@@ -64,6 +64,12 @@ clinic_directory:
   table:      fact_emr_treatment_out
   id_field:   organization_id
   name_field: organization_name
+  # ⚠️ 必须按时间取最新 —— 诊所改过名,源表里同一个 id 存着历次名字,且**旧名往往行数更多**:
+  #   273393ed… 「正大诊所」91,053 行(2016-07~2022-08,早停用)
+  #             「花旗医院」3,096 行(2023-06~2023-09,过渡)
+  #             「瑞尔齿科上海花旗诊所」42,646 行(2023-09~至今,现名)
+  #   66 家里 53 家都有多个名字。取 any()/最高频都会拿到旧名。
+  time_field: updated_date
 
 amount_unit:   yuan
 timezone:      Asia/Shanghai

apps/pac-service/src/cli/refresh-clinic-names.cli.ts

@@ -25,6 +25,11 @@ import { createClient } from '@clickhouse/client';
 import { AppModule } from '../app.module';
 import { PrismaService } from '../prisma/prisma.service';
 import { ColdImportManifestSchema } from '../modules/sync/cold-import/manifest.schema';
+import {
+  buildClinicNameQuery,
+  pickLatestNames,
+  type ClinicDirectoryConfig,
+} from '../modules/sync/cold-import/clinic-directory';
 import { disableSchedulersForCli } from './bootstrap-flags';
 
 // CLI 是短命进程,不需要 org-tree 启动预热(且会在 app.close() 时跟后台 warmAll 抢连接报噪音)。
@@ -37,7 +42,7 @@ function flag(name: string): string | undefined {
 
 async function deriveFromClickhouse(
   source: NonNullable<ReturnType<typeof ColdImportManifestSchema.parse>['sql_source']>,
-  dir: { table: string; id_field: string; name_field: string },
+  dir: ClinicDirectoryConfig,
   logger: Logger,
 ): Promise<Record<string, string>> {
   const url = process.env.DW_CLICKHOUSE_URL?.trim() || source.connection.url;
@@ -51,12 +56,15 @@ async function deriveFromClickhouse(
     const realTable = (() => {
       const sql = source.queries[dir.table];
       const m = sql?.match(/FROM\s+([\w.]+)/i);
-      return m ? m[1] : dir.table;
+      return m?.[1] ?? dir.table;
     })();
-    const q =
-      `SELECT ${dir.id_field} AS id, any(${dir.name_field}) AS name ` +
-      `FROM ${realTable} WHERE notEmpty(${dir.id_field}) AND notEmpty(${dir.name_field}) ` +
-      `GROUP BY ${dir.id_field}`;
+    if (!dir.time_field) {
+      logger.warn(
+        `clinic_directory 未声明 time_field —— 无法判断哪个名字是现名,改名过的诊所会拿到旧名。` +
+          `请在 manifest 补 time_field(如 updated_date)。`,
+      );
+    }
+    const q = buildClinicNameQuery(realTable, dir);
     logger.log(`CH 派生: ${q}`);
     const rs = await client.query({ query: q, format: 'JSONEachRow' });
     const rows = (await rs.json()) as Array<{ id: string; name: string }>;
@@ -71,7 +79,8 @@ async function deriveFromClickhouse(
 function deriveFromFile(
   manifestDir: string,
   tables: Array<{ table: string; file: string; format?: 'csv' | 'json' }>,
-  dir: { table: string; id_field: string; name_field: string },
+  dir: ClinicDirectoryConfig,
+  logger?: Logger,
 ): Record<string, string> {
   const t = tables.find((x) => x.table === dir.table);
   if (!t) throw new Error(`directory.table=${dir.table} 不在 tables[] 里`);
@@ -80,15 +89,12 @@ function deriveFromFile(
   const rows: Array<Record<string, unknown>> = t.file.toLowerCase().endsWith('.json')
     ? JSON.parse(raw)
     : parseCsv(raw, { columns: true, skip_empty_lines: true, trim: true });
-  const map: Record<string, string> = {};
-  for (const r of rows) {
-    const id = r[dir.id_field];
-    const name = r[dir.name_field];
-    if (id != null && String(id) !== '' && name != null && String(name) !== '') {
-      map[String(id)] = String(name);
-    }
+  if (!dir.time_field) {
+    logger?.warn(
+      `clinic_directory 未声明 time_field —— 无法判断哪个名字是现名,改名过的诊所会拿到旧名。`,
+    );
   }
-  return map;
+  return pickLatestNames(rows, dir);
 }
 
 async function main(): Promise<void> {
@@ -111,7 +117,7 @@ async function main(): Promise<void> {
 
   const clinicMap = manifest.sql_source
     ? await deriveFromClickhouse(manifest.sql_source, cd, logger)
-    : deriveFromFile(manifestDir, manifest.tables ?? [], cd);
+    : deriveFromFile(manifestDir, manifest.tables ?? [], cd, logger);
   const nc = Object.keys(clinicMap).length;
   logger.log(`派生诊所名 ${nc} 家`);
   if (nc === 0) {

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

+/**
+ * 诊所名字典派生 —— 从摄入源的 (id, name) 里挑出**现名**。
+ *
+ * 🔴 这里唯一要解决的问题:**诊所会改名,源表里同一个 id 存着历次名字。**
+ *   2026-09-03 生产实测(jvs-dw fact_emr_treatment_out,66 家诊所):
+ *     · **53 家有多个名字**(80%),不是个案
+ *     · 273393ed…  「正大诊所」91,053 行(2016-07~2022-08,早停用)
+ *                  「花旗医院」3,096 行(2023-06~2023-09,过渡)
+ *                  「瑞尔齿科上海花旗诊所」42,646 行(2023-09~至今,**现名**)
+ *     · c59bfc52…  6 个名字,其中两个带前导 `\t`
+ *
+ *   ⛔ `any(name)`(原实现)任取一个 —— 不确定,且大概率给旧名。
+ *   ⛔ "取最高频"同样错 —— 旧名用了 6 年,行数往往碾压现名。
+ *   ✅ 只能按时间取最新。
+ *
+ * 缺 `time_field` 时退化成"字典序最大":唯一的好处是**确定性**(同样输入同样输出,
+ * 不会跑一次一个名字),但**不保证是现名** —— 调用方必须告警。
+ */
+export interface ClinicDirectoryConfig {
+  table: string;
+  id_field: string;
+  name_field: string;
+  /// 判定"哪个名字是现名"的时间列(越大越新);字符串日期也可(字典序即时序)
+  time_field?: string;
+}
+
+/**
+ * ClickHouse 派生 SQL。
+ * - `trimBoth` 清掉源里带前导 `\t` 的脏值(华贸诊所实测有)
+ * - `argMax(name, time)` 取时间最大那行的名字
+ * - `ifNull(toString(time), '')` 兼容 Nullable 时间列(jvs-dw 的 updated_date 是 Nullable(String))
+ */
+export function buildClinicNameQuery(realTable: string, dir: ClinicDirectoryConfig): string {
+  const nameExpr = `trimBoth(${dir.name_field})`;
+  const pick = dir.time_field
+    ? `argMax(${nameExpr}, ifNull(toString(${dir.time_field}), ''))`
+    : `argMax(${nameExpr}, ${nameExpr})`;
+  return (
+    `SELECT ${dir.id_field} AS id, ${pick} AS name ` +
+    `FROM ${realTable} WHERE notEmpty(${dir.id_field}) AND notEmpty(${nameExpr}) ` +
+    `GROUP BY ${dir.id_field}`
+  );
+}
+
+/**
+ * 文件源派生 —— 与 ClickHouse 分支**同口径**。
+ * ⛔ 不能写成"遍历行、后者覆盖前者":行序是文件顺序,等于随机取名。
+ */
+export function pickLatestNames(
+  rows: Array<Record<string, unknown>>,
+  dir: ClinicDirectoryConfig,
+): Record<string, string> {
+  const best = new Map<string, { name: string; at: string }>();
+  for (const r of rows) {
+    const id = r[dir.id_field];
+    if (id == null || String(id) === '') continue;
+    const nm = r[dir.name_field] == null ? '' : String(r[dir.name_field]).trim();
+    if (!nm) continue;
+    // 无 time_field → 拿名字自身当序(字典序最大);有则按时间列
+    const at = dir.time_field ? String(r[dir.time_field] ?? '') : nm;
+    const cur = best.get(String(id));
+    if (!cur || at >= cur.at) best.set(String(id), { name: nm, at });
+  }
+  const out: Record<string, string> = {};
+  for (const [id, v] of best) out[id] = v.name;
+  return out;
+}

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

@@ -231,11 +231,19 @@ export const ColdImportManifestSchema = z
     /// `refresh-clinic-names` CLI 据此 SELECT DISTINCT 出 id→名,写 host.clinicNames;
     /// /auth/session 再合并进 dictionary.clinics 下发,前端显示名字不依赖登录传。
     /// sql_source 模式:table 是 CH 表名(或 queries 的 key);文件模式:table 对应 tables[].table。
+    /// 诊所名字典来源(id → 中文名),供 refresh-clinic-names 派生 host.clinicNames。
+    /// ⚠️ `time_field` 不是可有可无的:诊所会改名,同一个 id 在源表里存着**历次**名字。
+    ///   实测 jvs-dw 66 家里 **53 家有多个名字**(如 273393ed… 同时有「正大诊所」91,053 行
+    ///   〔2016~2022 旧名〕/「花旗医院」3,096 行 /「瑞尔齿科上海花旗诊所」42,646 行〔当前〕)。
+    ///   不按时间取最新,就会把早已停用的旧名当成现名 —— 而且旧名往往行数更多,
+    ///   "取最高频"同样错。缺 time_field 时只能退化成"最高频"并告警,结果不可信。
     clinic_directory: z
       .object({
         table: z.string().min(1),
         id_field: z.string().min(1),
         name_field: z.string().min(1),
+        /// 判定"哪个名字是现名"的时间列(越大越新)。字符串日期也可(字典序即时序)。
+        time_field: z.string().min(1).optional(),
       })
       .optional(),
 

apps/pac-service/tests/clinic-directory-latest-name.spec.ts

+/**
+ * 诊所名字典:必须取**现名**,不是任取 / 不是最高频。
+ *
+ * 🔴 起因(2026-09-03 生产实测):原实现 `any(organization_name)` 把
+ *   `273393edd04e4aa4afe9347bb8a2da21` 解析成「正大诊所」—— 那是 2022-08 就停用的旧名,
+ *   现名是「瑞尔齿科上海花旗诊所」。DW 里 66 家诊所有 **53 家**存着多个历史名字,
+ *   而且旧名用了六七年、行数往往碾压现名,所以"取最高频"跟 any() 一样错。
+ *
+ * 下面的样例数据直接取自生产 DW 的真实形态(含带前导 \t 的脏值)。
+ */
+import {
+  buildClinicNameQuery,
+  pickLatestNames,
+  type ClinicDirectoryConfig,
+} from '../src/modules/sync/cold-import/clinic-directory';
+
+const DIR: ClinicDirectoryConfig = {
+  table: 'fact_emr_treatment_out',
+  id_field: 'organization_id',
+  name_field: 'organization_name',
+  time_field: 'updated_date',
+};
+
+// 上海花旗:旧名行数(91,053)远超现名(42,646)—— 最高频口径会选错
+const SHANGHAI_CITI = [
+  ...rows('273393ed', '正大诊所', '2022-08-17 14:16:26', 3),
+  ...rows('273393ed', '花旗医院', '2023-09-07 15:01:58', 2),
+  ...rows('273393ed', '瑞尔齿科上海花旗诊所', '2026-09-03 08:55:18', 1),
+];
+function rows(id: string, name: string, at: string, n: number) {
+  return Array.from({ length: n }, () => ({
+    organization_id: id,
+    organization_name: name,
+    updated_date: at,
+  }));
+}
+
+describe('pickLatestNames —— 取现名而不是旧名', () => {
+  test('⭐ 旧名行数更多也要选最新的那个(上海花旗真实案例)', () => {
+    expect(pickLatestNames(SHANGHAI_CITI, DIR)).toEqual({
+      '273393ed': '瑞尔齿科上海花旗诊所',
+    });
+  });
+
+  test('行序打乱不影响结果(文件源的行序等于随机)', () => {
+    const shuffled = [...SHANGHAI_CITI].reverse();
+    expect(pickLatestNames(shuffled, DIR)['273393ed']).toBe('瑞尔齿科上海花旗诊所');
+  });
+
+  test('清掉源里的前导/尾随空白(华贸诊所实测有带 \\t 的行)', () => {
+    const r = pickLatestNames(
+      [
+        { organization_id: 'c59bfc52', organization_name: '华贸诊所', updated_date: '2020-01-01' },
+        { organization_id: 'c59bfc52', organization_name: '\t瑞尔齿科北京华贸诊所A诊区', updated_date: '2026-01-01' },
+      ],
+      DIR,
+    );
+    expect(r['c59bfc52']).toBe('瑞尔齿科北京华贸诊所A诊区');
+  });
+
+  test('空名 / 空 id / 全空白名 一律跳过,不写进字典', () => {
+    const r = pickLatestNames(
+      [
+        { organization_id: 'a', organization_name: '有效诊所', updated_date: '2020-01-01' },
+        { organization_id: 'a', organization_name: '   ', updated_date: '2026-01-01' },
+        { organization_id: '', organization_name: '无 id', updated_date: '2026-01-01' },
+        { organization_id: 'b', organization_name: null, updated_date: '2026-01-01' },
+      ],
+      DIR,
+    );
+    expect(r).toEqual({ a: '有效诊所' }); // 空白名不能顶掉有效名,b 不该出现
+  });
+
+  test('时间列缺值的行不会盖掉有时间的行', () => {
+    const r = pickLatestNames(
+      [
+        { organization_id: 'a', organization_name: '现名', updated_date: '2026-01-01' },
+        { organization_id: 'a', organization_name: '无时间旧名', updated_date: null },
+      ],
+      DIR,
+    );
+    expect(r['a']).toBe('现名');
+  });
+
+  test('未配 time_field → 退化成字典序最大(只保证确定性,不保证是现名)', () => {
+    const noTime: ClinicDirectoryConfig = { ...DIR, time_field: undefined };
+    const a = pickLatestNames(SHANGHAI_CITI, noTime);
+    const b = pickLatestNames([...SHANGHAI_CITI].reverse(), noTime);
+    expect(a).toEqual(b); // 同样输入同样输出,不会跑一次一个名字
+  });
+});
+
+describe('buildClinicNameQuery —— ClickHouse 侧同口径', () => {
+  test('⭐ 用 argMax 按时间取最新,⛔ 不再是 any()', () => {
+    const q = buildClinicNameQuery('dw_group.fact_emr_treatment_out', DIR);
+    expect(q).toContain("argMax(trimBoth(organization_name), ifNull(toString(updated_date), ''))");
+    expect(q).not.toMatch(/\bany\(/);
+  });
+
+  test('trimBoth 同时用在取值和非空判定上(否则纯 \\t 的行会被当成有效名)', () => {
+    const q = buildClinicNameQuery('t', DIR);
+    expect(q).toContain('notEmpty(trimBoth(organization_name))');
+  });
+
+  test('未配 time_field 也不退回 any()(保持确定性)', () => {
+    const q = buildClinicNameQuery('t', { ...DIR, time_field: undefined });
+    expect(q).not.toMatch(/\bany\(/);
+    expect(q).toContain('argMax(trimBoth(organization_name), trimBoth(organization_name))');
+  });
+});
+
+describe('manifest 必须声明 time_field(缺了就会解析成旧名)', () => {
+  const yaml = require('js-yaml');
+  const { readFileSync } = require('node:fs');
+  const { join } = require('node:path');
+  for (const host of ['jvs-dw', 'friday']) {
+    test(`${host}: clinic_directory.time_field 已声明`, () => {
+      const m = yaml.load(
+        readFileSync(join(__dirname, `../data/${host}/manifest.yaml`), 'utf-8'),
+      ) as { clinic_directory?: ClinicDirectoryConfig };
+      expect(m.clinic_directory?.time_field).toBeTruthy();
+    });
+  }
+});