Merge 5i5ya complaint ingestion into main

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

+# 标准反馈快照；有效投诉由标准性质+状态决定，不能整表/按评分猜投诉。
+canonical: complaint
+emits:
+  action: complaint_filed
+  subjectType: complaint
+  occurredAtField: filedAt
+  contentVersion: true
+primary:
+  table: complaint_rows
+  key: id
+# group_rows 已按 [brand,id] 聚合，不再取首行丢弃跟进。
+field_mapping:
+  externalId: id
+  patientExternalId: patient_id
+  clinicId: organization_id
+  filedAt: complain_date
+  createdAt: created_gmt_at
+  updatedAt: snapshot_changed_at
+  sourceUpdatedAt: updated_gmt_at
+  number: complain_number
+  content: complain_content
+  feedbackKind: complain_concerned
+  classificationKind: complain_classification
+  category: complain_concerned
+  channel: complain_channel
+  status: complain_type
+  sourceStatus: complain_type_name
+  sourceCategory: complain_classification_name
+  sourceConcern: complain_concerned_name
+  sourceChannel: complain_channel_name
+  relatedOrderExternalId: settlement_number
+  followUps: follow_ups
+# 未识别枚举保持未知；旧的“其他投诉”默认分类不强行盖过无不满/表扬。
+# 仍保留其原文供复核；明确服务/医疗投诉与表扬冲突时由通用 parser 标记未知。
+enum_mapping:
+  feedbackKind:
+    '1': complaint
+    '3': complaint
+    '4': complaint
+    '5': complaint
+    '6': complaint
+    '8': praise
+    '9': neutral
+    _default: unknown
+  classificationKind:
+    '1': complaint
+    '2': complaint
+    '4': praise
+    _default: unknown
+  category:
+    '1': service
+    '3': clinical
+    '4': billing
+    '5': waiting
+    '6': environment
+    _default: unknown
+  channel:
+    '1': hotline
+    '3': third_party
+    '4': sms_survey
+    '5': mini_program
+    '6': sales
+    _default: unknown
+  status:
+    '1': in_progress
+    '2': open
+    '4': resolved
+    '5': draft
+    _default: unknown

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

@@ -100,6 +100,8 @@ sql_source:
   incremental:
     per_query:
       fact_client_out:           { cursor_column: last_visit_time }
+      # 仅发现变化；明细查询不加游标，保证整单跟进完整。新表首轮补发现历史患者。
+      complaint_changes:         { cursor_column: changed_at, bootstrap_full: true }
       # 复杂病例:用 SELECT 里的 changed_at 别名(coalesce(updated,created));CH 支持 WHERE 引用别名。
       fact_complex_cases_out:    { cursor_column: changed_at }
       fact_appointment_out:      { cursor_column: updated_date }
@@ -133,12 +135,25 @@ sql_source:
       - fact_settlement_out
       - fact_settlement_mode_out
       - fact_complex_cases_out
+      - complaint_changes
 
   # SQL 最朴素化 — host(DW)给的数据范围就是 PAC 该消化的范围。
   # PAC 这边不预设诊所 / brand / 时间过滤,数据来什么就是什么。
   # dev 期的 cohort LIMIT 已去除 → 全量;只保留必要业务过滤(退费状态)。
   # cohort 用 (patient_id, brand) 复合键 — patient_id 跨 brand 撞车,需带 brand 才精确。
   queries:
+    # 投诉表混合满意度/表扬/投诉，完整读取变化后的状态，不在 SQL 过滤成仅未完成/负面。
+    # 字符串源时间均为北京时间；沿用 manifest.timezone 格式化游标，canonical 再归一到 UTC。
+    # 保留正常查询里的 changed_at 别名，供 cohort UNION 和空转探针共同复用。
+    complaint_changes: |
+      SELECT patient_id, brand,
+             greatest(ifNull(created_gmt_at, ''), ifNull(updated_gmt_at, ''), ifNull(follow_date, '')) AS changed_at
+      FROM dw_group.fact_complain_out
+      WHERE patient_id IS NOT NULL AND patient_id > 0 AND notEmpty(ifNull(brand, ''))
+    # 不配 per_query：被变化带进 cohort 的患者，拉全部投诉与跟进，而不是只有窗口内子行。
+    fact_complain_out: |
+      SELECT * FROM dw_group.fact_complain_out
+
     # ── 患者主档 ──
     # has_active_complex_case:宿主跟进闸(→ canonical hostFollowUpActive → patient_profiles)。
     #   DW「潜在治疗池」= FRIDAY 侧「复杂病例」:宿主自己在跟的患者,PAC 不再电话召回。
@@ -352,6 +367,36 @@ sql_source:
 # Layer A.5 transforms(SQL 干过的事搬这里,yaml 声明式)
 # ─────────────────────────────────────────────────────────────
 transforms:
+  # 投诉主单+跟进（源主字段冲突走拒绝举证，不随机匹配；空品牌不回落单命名空间）。
+  - kind: derive
+    input: fact_complain_out
+    output: fact_complain_out
+    fields:
+      follow_action:
+        op: recode
+        from: follow_type
+        default: unknown
+        map: { '转交': transferred, '完成': completed, '编辑': edited, '跟进': followed_up, '暂时搁置': paused }
+  - kind: group_rows
+    input: fact_complain_out
+    output: complaint_rows
+    by: [brand, id]
+    consistent_fields: [patient_id, organization_id, complain_number, complain_date, created_gmt_at, updated_gmt_at,
+      complain_content, complain_type, complain_type_name, complain_concerned, complain_concerned_name,
+      complain_classification, complain_classification_name, complain_channel, complain_channel_name, settlement_number]
+    max_fields:
+      snapshot_changed_at: [created_gmt_at, updated_gmt_at, follow_date]
+    collect:
+      follow_ups:
+        present_when: [follow_date, follow_type, follow_content, follow_user_id, follow_user_name]
+        fields:
+          occurredAt: follow_date
+          action: follow_action
+          sourceAction: follow_type
+          content: follow_content
+          actorId: follow_user_id
+          actorName: follow_user_name
+
   # ── A. EMR 主表 → 只留自由文本(剥离 JSON 字段) ──
   - kind: project
     input: fact_emr_treatment_out
@@ -1004,6 +1049,7 @@ transforms:
 # PAC 写的 assembler 配置(每个 canonical resource 一份)
 # ─────────────────────────────────────────────────────────────
 assemblers:
+  - { file: assemblers/complaint.yaml }
   - { file: assemblers/patient.yaml }
   - { file: assemblers/patient_relation.yaml }   # 联系人/亲属边(referee → PatientRelation upsert)
   - { file: assemblers/patient_return_visit.yaml }  # 诊所回访(展示用,全瑞尔 → PatientReturnVisit upsert)

apps/pac-service/src/modules/sync/assembler/field-mapper.ts

@@ -188,6 +188,16 @@ export function normalizeCanonical(
     }
   }
 
+  // 嵌套业务时间也按宿主时区解释，禁止跟进裸时间依赖 Node 进程时区。
+  for (const { array, field } of meta.datetimeArrayFields ?? []) {
+    const items = coerceArray(out[array]);
+    if (items) out[array] = items.map(item => {
+      if (!item || typeof item !== 'object') return item;
+      const row = item as Record<string, unknown>;
+      return typeof row[field] === 'string'
+        ? { ...row, [field]: normalizeDatetime(row[field] as string, ctx.timezone) } : item;
+    });
+  }
   return out;
 }
 

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

@@ -76,7 +76,7 @@ export function resolveOnlyPatientKeys(): OnlyPatientKey[] {
 /// W4 末:DW 增量配置(per table cursor)
 export interface IncrementalConfig {
   /// per query 配置:cursor 列名(host 表那边的字段)+ 上次 cursor 值(从 sync_logs 读)
-  perQuery: Record<string, { cursorColumn: string; cursorValue: string | null }>;
+  perQuery: Record<string, { cursorColumn: string; cursorValue: string | null; bootstrapFull?: boolean }>;
   /// 跑完后返回新的 cursor 值(每张表的 max)
   cursorAdvances?: Record<string, string>;
 }
@@ -223,11 +223,8 @@ export class ClickHouseSourceService {
           //   反向结算 is_refund=1、无就诊记录患者 last_visit_time NULL)→ 真实拉取正确地不拉,
           //   探针却数到 → 误报"增量空转"CRITICAL(2026-07-25 实测:settlement 正/退表各 5788
           //   完全相等,正是这批全为退费单的铁证)。
-          const clauses = [
-            `${cfg.cursorColumn} > '${cfg.cursorValue.replace(/'/g, "''")}'`,
-            ...this.extractBusinessFilters(sql),
-          ];
-          const q = `SELECT count(*) AS c FROM ${m[1]} WHERE ${clauses.join(' AND ')}`;
+          // 包裹真实增量查询，保留 changed_at 等 SELECT 派生别名，避免探针直查物理表报未知列。
+          const q = `SELECT count(*) AS c FROM (${this.injectIncrementalCursor(sql, cfg.cursorColumn, cfg.cursorValue)})`;
           const rs = await client.query({ query: q, format: 'JSONEachRow' });
           const rows = (await rs.json()) as Array<{ c: string | number }>;
           out[tableName] = Number(rows[0]?.c ?? 0);
@@ -483,7 +480,7 @@ export class ClickHouseSourceService {
       const incBranches: string[] = [];
       if (incremental) {
         for (const [tbl, cfg] of Object.entries(incremental.perQuery)) {
-          if (!cfg.cursorColumn || !cfg.cursorValue) continue;
+          if (!cfg.cursorColumn || (!cfg.cursorValue && !cfg.bootstrapFull)) continue;
           const q = source.queries[tbl];
           const keyCols = tenant_key_column
             ? `${patient_key_column}, ${tenant_key_column}`
@@ -505,7 +502,8 @@ export class ClickHouseSourceService {
           }
           if (!branch) {
             const fqn = (q && this.splitSelectFrom(q)?.fromTable) ?? tbl;
-            branch = `SELECT ${keyCols} FROM ${fqn} WHERE ${cfg.cursorColumn} > '${cfg.cursorValue.replace(/'/g, "''")}'`;
+            branch = `SELECT ${keyCols} FROM ${fqn}` + (cfg.cursorValue
+              ? ` WHERE ${cfg.cursorColumn} > '${cfg.cursorValue.replace(/'/g, "''")}'` : '');
           }
           incBranches.push(branch);
         }

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

@@ -971,6 +971,7 @@ export class ColdImportService {
             table,
             {
               cursorColumn: cfg.cursor_column,
+              bootstrapFull: cfg.bootstrap_full,
               // ⚠️ 必须规范化:DW 的 cursor 列是 String(如 updated_date: Nullable(String),
               // 值形如 'YYYY-MM-DD HH:mm:ss')→ WHERE 比较是字典序。游标存的是 run_start ISO
               // ('2026-06-09T03:30:00.012Z'),第 11 字符 'T'(0x54) > ' '(0x20) → 所有真实行

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

@@ -66,6 +66,8 @@ export const IncrementalDeclSchema = z.object({
     z.string(),
     z.object({
       cursor_column: z.string().min(1),
+      // 新资源无水位时主动发现历史患者；默认保持旧资源行为。
+      bootstrap_full: z.boolean().optional(),
     }),
   ),
 });

apps/pac-service/src/modules/sync/pipeline/fact-content-schemas.ts

@@ -17,6 +17,7 @@
  */
 import { z } from 'zod';
 import {
+  ComplaintStatusSchema, ComplaintFeedbackKindSchema, ComplaintCategorySchema, ComplaintChannelSchema,
   ClinicalSignalsSchema,
   FactType,
   PACDiagnosisCodeSchema,
@@ -445,7 +446,11 @@ const ComplaintRecordContent = z
   .object({
     complaint_external_id: z.string().min(1),
     severity: z.enum(['low', 'medium', 'high']).nullable().optional().default(null),
-    status: z.enum(['filed', 'in_progress', 'resolved']).nullable().optional().default(null),
+    status: ComplaintStatusSchema.nullable().optional().default(null),
+    feedback_kind: ComplaintFeedbackKindSchema.optional(),
+    category: ComplaintCategorySchema.optional(),
+    channel: ComplaintChannelSchema.optional(),
+    signal_eligible: z.boolean().optional(),
     content: nullableString(),
   })
   .passthrough();

apps/pac-service/src/modules/sync/pipeline/parsers/complaint.parser.ts

-import { Injectable, Logger } from '@nestjs/common';
-import { Action, FactKind, FactStatus, FactType } from '@pac/types';
-import type { FactDraft, Parser, ParserContext } from './parser.interface';
+import { Injectable } from '@nestjs/common';
+import { Action, FactKind, FactStatus, FactType, ComplaintCanonicalSchema } from '@pac/types';
+import type { FactDraft, FactSnapshotScope, Parser, ParserContext } from './parser.interface';
 
-/**
- * ComplaintParser — `complaint_filed` / `complaint_resolved` 解析器
- *
- * 产 1 个 `complaint_record`(actual)— 投诉事实。
- * 同 subject_id 在 filed → resolved 两次 transaction 间走版本流(supersede)。
- *
- * 合规重要性:Persona.do_not_contact_status 应消费 active complaint_record 触发 DNC。
- */
+/** 投诉标准快照；只负责事实，不设置免打扰、不改变召回规则。 */
 @Injectable()
 export class ComplaintParser implements Parser {
   readonly name = Action.COMPLAINT_FILED;
-  private readonly logger = new Logger(ComplaintParser.name);
 
-  parse(ctx: ParserContext): FactDraft[] {
-    const c = ctx.canonicalRow;
-    const externalId = String(c.externalId ?? ctx.transaction.subjectId).trim();
-    if (!externalId) {
-      this.logger.warn(`complaint parser: missing externalId; tx=${ctx.transaction.id};skip`);
-      return [];
-    }
-
-    const filedAt = c.filedAt ? new Date(c.filedAt as string) : null;
-    const resolved = ctx.transaction.action === Action.COMPLAINT_RESOLVED;
-    const resolvedAt = c.resolvedAt ? new Date(c.resolvedAt as string) : null;
-    const hostStatus = String(c.status ?? (resolved ? 'resolved' : 'open')).toLowerCase();
-    const factStatus = hostStatus === 'resolved' ? FactStatus.FULFILLED : FactStatus.ACTIVE;
+  snapshotScope(ctx: ParserContext): FactSnapshotScope {
+    const c = ComplaintCanonicalSchema.parse(ctx.canonicalRow);
+    return { anchorSubjectId: `${FactType.COMPLAINT_RECORD}:${c.externalId}`,
+      memberType: FactType.COMPLAINT_RECORD, contentKey: 'complaint_external_id',
+      externalId: c.externalId, replace: false };
+  }
 
-    return [
-      {
-        subjectId: `${FactType.COMPLAINT_RECORD}:${externalId}`,
-        kind: FactKind.ACTUAL,
-        type: FactType.COMPLAINT_RECORD,
-        status: factStatus,
-        clinicId: ctx.transaction.clinicId,
-        occurredAt: filedAt,
-        title: `投诉 ${externalId}${resolved ? '(已结案)' : ''}`,
-        summary: (c.content as string | undefined) ?? null,
-        content: {
-          complaint_external_id: externalId,
-          host_status: hostStatus,
-          category: c.category ?? null,
-          content: c.content ?? null,
-          related_treatment_subject_id: c.relatedTreatmentSubjectId ?? null,
-          resolved_at: resolvedAt?.toISOString() ?? null,
-        },
+  parse(ctx: ParserContext): FactDraft[] {
+    const c = ComplaintCanonicalSchema.parse(ctx.canonicalRow);
+    const classificationConflict = c.classificationKind !== 'unknown' && c.feedbackKind !== 'unknown'
+      && c.classificationKind !== c.feedbackKind;
+    const kind = classificationConflict ? 'unknown' : c.feedbackKind;
+    const eligible = kind === 'complaint' && ['open', 'in_progress', 'resolved'].includes(c.status);
+    const followUps = c.followUps.map(f => ({ ...f,
+      occurredAt: f.occurredAt ? new Date(f.occurredAt).toISOString() : null,
+    })).sort((a, b) => JSON.stringify(a).localeCompare(JSON.stringify(b)));
+    const completionTimes = followUps.filter(f => f.action === 'completed' && f.occurredAt)
+      .map(f => f.occurredAt!);
+    if (c.resolvedAt) completionTimes.push(new Date(c.resolvedAt).toISOString());
+    const resolvedAt = c.status === 'resolved' ? completionTimes.sort().at(-1) ?? null : null;
+    // 非投诉/未知/草稿保留无效快照：新分类能替换旧有效投诉，旧报文重试不能复活它。
+    // ⛔ 不把这些快照当有效投诉统计；消费时检查 signal_eligible 与最新版本。
+    return [{
+      subjectId: `${FactType.COMPLAINT_RECORD}:${c.externalId}`,
+      kind: FactKind.ACTUAL, type: FactType.COMPLAINT_RECORD,
+      status: !eligible ? FactStatus.INVALIDATED
+        : c.status === 'resolved' ? FactStatus.FULFILLED : FactStatus.ACTIVE,
+      clinicId: c.clinicId, occurredAt: new Date(c.filedAt),
+      title: eligible ? `投诉${c.status === 'resolved' ? '(已完成)' : ''}` : '患者反馈（非有效投诉）',
+      summary: c.content ?? null,
+      content: {
+        complaint_external_id: c.externalId, number: c.number ?? null,
+        feedback_kind: kind, status: c.status, category: c.category, channel: c.channel,
+        signal_eligible: eligible,
+        review_reason: classificationConflict ? 'classification_conflict'
+          : kind === 'unknown' ? 'unknown_feedback_kind' : c.status === 'unknown' ? 'unknown_status' : null,
+        content: c.content ?? null, severity: null,
+        related_treatment_subject_id: c.relatedTreatmentSubjectId ?? null,
+        related_order_external_id: c.relatedOrderExternalId ?? null,
+        resolved_at: resolvedAt, follow_ups: followUps,
+        source_updated_at: c.sourceUpdatedAt ? new Date(c.sourceUpdatedAt).toISOString() : null,
+        host_status: c.sourceStatus ?? c.status,
+        source_category: c.sourceCategory ?? null, source_concern: c.sourceConcern ?? null,
+        source_channel: c.sourceChannel ?? null,
       },
-    ];
+    }];
   }
 }

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

@@ -122,8 +122,8 @@ function evalExpr(expr: DeriveExpr, row: Row): unknown {
     }
     case 'recode': {
       const v = row[expr.from];
-      if (typeof v !== 'string') return v ?? null;
-      return Object.prototype.hasOwnProperty.call(expr.map, v) ? expr.map[v] : v;
+      if (typeof v !== 'string') return expr.default ?? v ?? null;
+      return Object.prototype.hasOwnProperty.call(expr.map, v) ? expr.map[v] : (expr.default ?? v);
     }
   }
 }

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

+import { createHash } from 'node:crypto';
+import type { GroupRowsOp } from '../transforms.schema';
+import { SOURCE_ROW_KEY, isEmptyText, type Row } from '../row';
+
+export const GROUP_ERROR = '__group_error';
+const ENVELOPE = '__pac_group_rows';
+const SNAPSHOT = '__group_snapshot';
+
+/** 原文是多行快照时，离线重放恢复原输入表，不把一条跟进冒充整单。 */
+export function unpackGroupRows(table: string, row: Row): Row[] {
+  const envelope = row[ENVELOPE] as { table?: string; rows?: Row[] } | undefined;
+  if (envelope?.table !== table || !Array.isArray(envelope.rows)) return [row];
+  // 每个封套只算一次短指纹，避免把完整原文重复拼进每条跟进的分组键。
+  const snapshot = createHash('sha256').update(stable(envelope)).digest('hex');
+  return envelope.rows.map(raw => ({ ...raw, [SOURCE_ROW_KEY]: raw, [SNAPSHOT]: snapshot }));
+}
+
+function stable(value: unknown): string {
+  return JSON.stringify(value, (_key, v) => v && typeof v === 'object' && !Array.isArray(v)
+    ? Object.fromEntries(Object.keys(v).sort().map(k => [k, v[k]])) : v);
+}
+function uniqueSorted(rows: Row[]): Row[] {
+  return [...new Map(rows.map(r => [stable(r), r])).entries()]
+    .sort(([a], [b]) => a < b ? -1 : a > b ? 1 : 0).map(([, r]) => r);
+}
+
+/** 仅聚合配置指定的字段；身份缺失/主体矛盾送 assembler 的拒绝举证。 */
+export function runGroupRows(op: GroupRowsOp, input: Row[]): Row[] {
+  const groups = new Map<string, Row[]>();
+  input.forEach((row, i) => {
+    const keys = op.by.map(k => row[k]);
+    const key = keys.some(isEmptyText) ? `invalid:${i}` : stable([row[SNAPSHOT] ?? null, ...keys.map(String)]);
+    const rows = groups.get(key) ?? [];
+    rows.push(row);
+    groups.set(key, rows);
+  });
+  return [...groups.values()].map(rows => {
+    const raw = uniqueSorted(rows.map(row => (row[SOURCE_ROW_KEY] ?? row) as Row));
+    const first = rows[0]!;
+    const out: Row = Object.fromEntries([...new Set([...op.by, ...op.consistent_fields])]
+      .map(k => [k, first[k] ?? null]));
+    const missing = op.by.filter(k => isEmptyText(first[k]));
+    const conflicts = op.consistent_fields.filter(k => new Set(rows.map(row => stable(row[k] ?? null))).size > 1);
+    if (missing.length || conflicts.length) {
+      out[GROUP_ERROR] = `group_rows: missing keys [${missing.join(',')}]; conflicting fields [${conflicts.join(',')}]`;
+    }
+    for (const [target, fields] of Object.entries(op.max_fields)) {
+      const values = rows.flatMap(row => fields.map(f => row[f])).filter(v => !isEmptyText(v));
+      // 此算子只比较同格式源时间字符串，不猜测时区/混合格式。
+      if (values.some(v => typeof v !== 'string' || !/^\d{4}-\d\d-\d\d[ T]\d\d:\d\d:\d\d(?:\.\d+)?$/.test(v))) {
+        out[GROUP_ERROR] = `group_rows: invalid comparable datetime [${target}]`;
+      }
+      out[target] = values.length ? values.map(String).sort().at(-1)! : null;
+    }
+    for (const [target, collect] of Object.entries(op.collect)) {
+      out[target] = uniqueSorted(rows.filter(row => collect.present_when.some(f => !isEmptyText(row[f])))
+        .map(row => Object.fromEntries(Object.entries(collect.fields).map(([k, f]) => [k, row[f] ?? null]))));
+    }
+    // 顶层保留原始主体字段供租户/命名空间解析，完整原行放有标识的证据封套。
+    out[SOURCE_ROW_KEY] = { ...raw[0], [ENVELOPE]: { table: op.input, rows: raw } };
+    return out;
+  });
+}

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

 import { Injectable, Logger } from '@nestjs/common';
+import { runGroupRows, unpackGroupRows } from './operators/group-rows.op';
 import { runProject } from './operators/project.op';
 import { runSplitJsonArray } from './operators/split-json-array.op';
 import { runDerive } from './operators/derive.op';
@@ -36,7 +37,7 @@ export class TransformEngine {
     const tables: TableMap = {};
     // 入口标记:每个源行带上自身的 __source_row 引用(深拷贝避免后续 operator 修改污染)
     for (const [name, rows] of Object.entries(opts.tables)) {
-      tables[name] = rows.map((row) => {
+      tables[name] = rows.flatMap(row => unpackGroupRows(name, row)).map((row) => {
         // 已有 __source_row 的不覆盖(允许 caller 提前注入,比如 Push 路径)
         if (SOURCE_ROW_KEY in row) return row;
         return { ...row, [SOURCE_ROW_KEY]: row };
@@ -63,6 +64,12 @@ export class TransformEngine {
       }
 
       switch (op.kind) {
+        case 'group_rows': {
+          const out = runGroupRows(op, input);
+          tables[op.output] = out;
+          this.logger.log(`[transform/group_rows] "${inputName}"(${input.length}) → "${op.output}"(${out.length})`);
+          break;
+        }
         case 'project': {
           const out = runProject(op, input);
           tables[op.output] = out;

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

@@ -190,6 +190,7 @@ export const DeriveCoalesceSchema = z.object({
 ///   { from: std_code, map: { 'K07.303': 'K03' } } → K07.303 改 K03,再交 substring 截短;其余码不变。
 export const DeriveRecodeSchema = z.object({
   op: z.literal('recode'),
+  default: z.string().optional(),
   from: z.string().min(1),
   map: z.record(z.string(), z.string()).describe('原值 → 目标值;未命中则透传原值'),
 });
@@ -422,6 +423,22 @@ export type LookupOp = z.infer<typeof LookupOpSchema>;
 // operator union + transforms[] 顶层
 // ─────────────────────────────────────────────────────────
 
+/** 一对多展开表还原稳定主体快照；字段/身份均由宿主配置声明。 */
+export const GroupRowsOpSchema = z.object({
+  kind: z.literal('group_rows'),
+  input: tableNameSchema,
+  output: tableNameSchema,
+  by: z.array(z.string().min(1)).min(1),
+  consistent_fields: z.array(z.string().min(1)).min(1),
+  // 同格式、同时区的源时间字符串求最大；时区归一在 canonical 入口完成。
+  max_fields: z.record(z.string(), z.array(z.string().min(1)).min(1)).default({}),
+  collect: z.record(z.string(), z.object({
+    fields: z.record(z.string(), z.string()),
+    present_when: z.array(z.string().min(1)).min(1),
+  })).default({}),
+});
+export type GroupRowsOp = z.infer<typeof GroupRowsOpSchema>;
+
 export const TransformOpSchema = z.discriminatedUnion('kind', [
   ProjectOpSchema,
   SplitJsonArrayOpSchema,
@@ -431,6 +448,7 @@ export const TransformOpSchema = z.discriminatedUnion('kind', [
   FilterOpSchema,
   UnionOpSchema,
   LookupOpSchema,
+  GroupRowsOpSchema,
 ]);
 export type TransformOp = z.infer<typeof TransformOpSchema>;
 

apps/pac-service/tests/canonical-fact-layer.spec.ts

@@ -23,6 +23,7 @@ import {
   PACImageModalities,
   FDI_TOOTH_POSITIONS,
   ImageAnalysisCanonicalSchema,
+  ComplaintStatusSchema, ComplaintFeedbackKindSchema, ComplaintCategorySchema, ComplaintChannelSchema,
 } from '@pac/types';
 import { FactContentSchemas } from '../src/modules/sync/pipeline/fact-content-schemas';
 import {
@@ -194,6 +195,19 @@ describe('canonical-fact-layer | 闸 4:yaml enum_mapping 目标 ∈ canonical-co
         });
         continue;
       }
+      // 同名字段按资源契约校验：投诉的 category/status 不是治疗类别/治疗状态。
+      if (config.canonical === 'complaint') {
+        const complaintEnums: Record<string, readonly string[]> = {
+          status: ComplaintStatusSchema.options, category: ComplaintCategorySchema.options,
+          channel: ComplaintChannelSchema.options, feedbackKind: ComplaintFeedbackKindSchema.options,
+          classificationKind: ComplaintFeedbackKindSchema.options,
+        };
+        const options = complaintEnums[canonicalField];
+        if (options) test(`${host}/${file}:complaint.${canonicalField} 及兜底值符合标准枚举`, () => {
+          for (const target of Object.values(valueMap)) expect(options).toContain(target);
+        });
+        continue;
+      }
       const allowed = canonicalField === 'status' && config.canonical === 'image_analysis'
         ? new Set<string>(ImageAnalysisCanonicalSchema.shape.status.options)
         : closedSets[canonicalField];

apps/pac-service/tests/jvs-complaint-incremental.spec.ts

+import { readFileSync, writeFileSync } from 'node:fs';
+import { join } from 'node:path';
+import * as yaml from 'js-yaml';
+import { Logger } from '@nestjs/common';
+import { createClient } from '@clickhouse/client';
+import { ColdImportManifestSchema } from '../src/modules/sync/cold-import/manifest.schema';
+import { ClickHouseSourceService, type IncrementalConfig } from '../src/modules/sync/cold-import/clickhouse-source.service';
+
+jest.mock('@clickhouse/client', () => ({ createClient: jest.fn() }));
+const manifest = ColdImportManifestSchema.parse(yaml.load(readFileSync(join(__dirname, '../data/jvs-dw/manifest.yaml'), 'utf8')));
+const source = { ...manifest.sql_source!, connection: { ...manifest.sql_source!.connection, password_env: 'PAC_TEST_COMPLAINT_CH_PASSWORD' } };
+beforeAll(() => { process.env.PAC_TEST_COMPLAINT_CH_PASSWORD = 'mock-only'; });
+afterAll(() => { delete process.env.PAC_TEST_COMPLAINT_CH_PASSWORD; });
+const svc = new ClickHouseSourceService();
+let queries: string[] = [];
+const cursor = '2026-10-16 22:15:00'; // 对应 UTC 2026-10-16T14:15:00Z，调用方已按源时区回转
+beforeEach(() => {
+  Logger.overrideLogger(false); queries = [];
+  (createClient as jest.Mock).mockReturnValue({
+    query: jest.fn(async ({ query }: { query: string }) => {
+      queries.push(query);
+      return { json: async () => query.startsWith('SELECT count(*)') ? [{ c: '3' }] : [] };
+    }), close: jest.fn(async () => undefined),
+  });
+});
+function increment(value: string | null): IncrementalConfig {
+  return { perQuery: { complaint_changes: { cursorColumn: 'changed_at', cursorValue: value, bootstrapFull: true } } };
+}
+
+test('已有宿主新增投诉资源：无游标时扫描历史变化患者，其他资源无游标不受影响', async () => {
+  await svc.listPatientPairs(source, increment(null));
+  expect(queries[0]).toContain('fact_complain_out');
+  expect(queries[0]).toContain('patient_id, brand');
+  expect(queries[0]).not.toContain('changed_at >');
+  expect(queries[0]).toContain('patient_id > 0');
+  const noBootstrap = increment(null); noBootstrap.perQuery.complaint_changes!.bootstrapFull = false;
+  queries = []; await svc.listPatientPairs(source, noBootstrap);
+  expect(queries[0]).not.toContain('fact_complain_out');
+});
+
+test('日常增量同时捕捉创建/主单更新/跟进更新，保留源时区游标', async () => {
+  await svc.listPatientPairs(source, increment(cursor));
+  expect(queries[0]).toContain("greatest(ifNull(created_gmt_at, ''), ifNull(updated_gmt_at, ''), ifNull(follow_date, '')) AS changed_at");
+  expect(queries[0]).toContain(`changed_at > '${cursor}'`);
+  expect(queries[0]).toContain('notEmpty');
+});
+
+test('发现变化与完整读取分开：跟进快照不被增量窗口切断；按配置品牌+患者圈定', async () => {
+  const subset = { ...source, queries: { complaint_changes: source.queries.complaint_changes!, fact_complain_out: source.queries.fact_complain_out! } };
+  await svc.loadTablesForCohort(subset, [{ key: '123', tenant: '瑞尔' }], increment(cursor));
+  const full = queries.find(q => q.includes('SELECT *'))!;
+  expect(full).toContain("(patient_id, brand) IN (('123', '瑞尔'))");
+  expect(full).not.toContain('changed_at >');
+  expect(full).not.toContain('follow_date >');
+  expect(queries.find(q => q.includes('AS changed_at'))).toContain(`changed_at > '${cursor}'`);
+});
+
+test('空转探针复用派生游标查询，不把 changed_at 当物理列', async () => {
+  expect(await svc.probeNewRowCounts(source, increment(cursor).perQuery)).toEqual({ complaint_changes: 3 });
+  expect(queries[0]).toMatch(/^SELECT count\(\*\) AS c FROM \(SELECT/);
+  expect(queries[0]).toContain('AS changed_at');
+  expect(queries[0]).toContain('patient_id > 0');
+});
+
+(process.env.PAC_COMPLAINT_SQL_EVIDENCE && process.env.PAC_COMPLAINT_REAL_FIXTURE ? test : test.skip)(
+  '导出实际生成SQL供独立只读数仓验证（仅本地私有路径）', async () => {
+    const raw = JSON.parse(readFileSync(process.env.PAC_COMPLAINT_REAL_FIXTURE!, 'utf8')) as Record<string, unknown>[];
+    const valid = raw.filter(r => r.patient_id && r.follow_date && String(r.follow_date) > String(r.updated_gmt_at ?? ''));
+    const chosen = valid[0]!;
+    const subset = { ...source, queries: { complaint_changes: source.queries.complaint_changes!, fact_complain_out: source.queries.fact_complain_out! } };
+    const mark = String(chosen.updated_gmt_at ?? chosen.created_gmt_at);
+    await svc.listPatientPairs(subset, increment(null));
+    const bootstrap = queries[0]; queries = [];
+    await svc.loadTablesForCohort(subset, [{ key: String(chosen.patient_id), tenant: String(chosen.brand) }], increment(mark));
+    const cohort = queries.slice(); queries = [];
+    await svc.probeNewRowCounts(subset, increment(mark).perQuery);
+    writeFileSync(process.env.PAC_COMPLAINT_SQL_EVIDENCE!, JSON.stringify({ bootstrap, cohort, probe: queries[0],
+      chosen: { id: chosen.id, brand: chosen.brand, patient_id: chosen.patient_id }, cursor: mark }), { mode: 0o600 });
+  });

docs/qa/2026-10-18-jvs-complaint-clinics.csv

+品牌,源诊所ID,源诊所名称,PAC字典名称,反馈单总数,负面关注点候选,候选未完成,表扬,无不满,关注点未分类,全类缺患者ID,候选草稿,候选非草稿且患者可关联源主档,候选可关联当前生产PAC患者,候选未完成可关联当前生产PAC患者,最近反馈时间
+瑞尔,bcda42d627dc4fe09eb248e21f9fd2f7,瑞尔齿科深圳诺德诊所,瑞尔齿科深圳诺德诊所,399,92,1,33,274,0,7,2,84,37,0,2026-09-03 16:41:41.000
+瑞尔,ceed7716d1cc40c8a6392fcb8928944a,瑞尔齿科厦门国际银行大厦诊所,瑞尔齿科厦门国际银行大厦诊所,139,49,0,5,77,8,0,0,49,27,0,2026-10-16 17:28:57.000
+瑞尔,ff90434c695f409d8f0fdfb35a5a6b76,瑞尔齿科深圳华侨城诊所,瑞尔齿科深圳华侨城诊所,171,43,0,27,100,1,1,0,42,23,0,2026-09-13 12:38:25.000
+瑞尔,031f22411ab94afea8355ef42264030f,瑞尔齿科深圳欢乐海岸诊所,瑞尔齿科深圳欢乐海岸诊所,376,42,0,52,282,0,2,1,40,20,0,2026-10-16 09:01:40.000
+瑞尔,2523178e193346c883118718672eeed8,瑞尔齿科广州甲子花园诊所,瑞尔齿科广州甲子花园诊所,112,38,3,1,73,0,4,0,34,12,0,2026-01-27 09:04:00.000
+瑞尔,ada5fcfe2a234eab9242d6d5d8dc0166,瑞尔齿科广州高德诊所,瑞尔齿科广州高德诊所,308,38,3,23,196,51,1,0,37,16,2,2026-10-16 12:57:05.000
+瑞尔,d53e735b28f944aa8770f24ed7ff8803,地王诊所,瑞尔齿科深圳地王诊所,184,37,0,10,137,0,6,0,32,16,0,2024-11-16 20:03:01.000
+瑞尔,d063b969850f461192f443a7f6a58ad0,卓越世纪诊所,卓越世纪诊所,147,26,0,12,109,0,0,0,26,13,0,2022-11-05 22:16:41.000
+瑞尔,b6f0dc73b6c240d98ac8e00f66ac3688,瑞尔齿科深圳南山恒裕中心诊所,瑞尔齿科深圳南山恒裕中心诊所,171,23,0,26,121,1,0,0,23,7,0,2026-10-17 19:38:01.000
+瑞尔,28cbf0a2e8484a75806c2246f896ad43,猎德诊所,瑞尔齿科广州猎德诊所,130,21,0,14,95,0,0,0,21,5,0,2022-12-12 16:00:21.000
+瑞尔,8301dc0da8274b1ba0a389bb0e736224,中石化诊所,中石化诊所,129,20,0,4,105,0,1,0,19,6,0,2024-08-14 15:13:25.000
+瑞泰,4c13b9c33330401fba469d871aaab6ec,成都瑞泰口腔医院武侯总院,,1495,276,0,232,983,4,41,0,244,45,0,2026-10-14 11:13:14.000
+瑞泰,1e9a41cd8c644859a157f70a7105bf7b,长沙瑞泰口腔科尔雅定王台医院,,717,141,0,138,437,1,20,0,122,13,0,2026-04-18 12:30:41.000
+瑞泰,484060c2ea1e4ead943cb731a746d20f,重庆瑞泰口腔医院,,1324,129,1,826,344,25,14,0,115,21,0,2026-10-12 15:16:43.000
+瑞泰,0f18cf193b0147f487467ac152968dab,西安瑞泰口腔尔仓医院,,674,113,0,142,416,3,12,0,101,10,0,2026-10-17 17:00:34.000
+瑞泰,38803ab6a9dc42f495390075b100cbe3,上海瑞泰口腔前滩门诊,,443,100,4,64,273,6,21,0,79,15,0,2026-09-27 15:07:26.000
+瑞泰,95eb5789205843c682f97d154534e374,西安瑞泰口腔雁塔医院,,504,95,0,140,269,0,16,0,79,1,0,2026-05-31 16:54:44.000
+瑞泰,13b0035726f742888274f3e30fffed2d,重庆瑞泰口腔江北门诊,,807,91,0,418,297,1,9,0,82,13,0,2026-10-17 16:15:13.000
+瑞泰,39d283de8ad84c3eb6bf6b5cbf26318e,青岛瑞泰口腔市北门诊,,350,83,0,55,212,0,16,0,69,4,0,2026-09-05 18:22:43.000
+瑞泰,9e256b519d1e4beea34400616387be89,长沙瑞泰科尔雅口腔医院中庭分院,,318,80,0,21,209,8,9,0,72,7,0,2026-08-22 09:27:39.000
+瑞泰,491cb4ee54c84e5a97f1c711f258586f,重庆瑞泰久悦口腔万象城门诊,,661,76,0,405,179,1,5,0,72,21,0,2023-11-26 11:00:20.000
+瑞泰,5d8b66cdd71c4806938e114007ac9f5b,重庆瑞泰口腔南方花园门诊,,596,74,0,336,186,0,2,0,72,11,0,2025-07-23 11:21:46.000
+瑞泰,70dbe1e5728843129c970805fcfd9249,重庆瑞泰口腔南CBD门诊,,454,72,1,144,237,1,10,0,63,10,0,2026-10-16 12:05:25.000
+瑞泰,4839f58bc8364da4b49ca1f3c51b4d7f,重庆瑞泰口腔爱琴海门诊,,339,69,0,135,135,0,11,0,59,12,0,2025-07-27 08:41:27.000
+瑞泰,2f44ac9c1f04425796d0af892e361733,北京瑞泰口腔医院工体分院,,278,66,0,99,112,1,0,0,66,6,0,2026-08-26 09:32:03.000
+瑞泰,235d58fd98ce4bcbb85cf91ad9055d37,成都瑞泰口腔医院贝森分院,,503,64,0,139,300,0,12,0,52,6,0,2024-03-29 09:05:08.000
+瑞泰,f5180b97b6de4705a6e5b1fb619dbfff,大连瑞泰口腔葆嘉门诊,,212,62,1,68,80,2,10,0,52,9,0,2026-10-12 11:00:53.000
+瑞泰,f464495631cf4c4ea3136a38480a017c,北京瑞泰口腔医院望京分院,,257,62,0,68,127,0,2,0,61,6,0,2026-09-28 11:28:32.000
+瑞泰,ad4a4eb7e36d4cadb5a2b0fb82250767,重庆瑞泰口腔龙湖门诊,,757,61,0,289,407,0,6,0,56,6,0,2026-04-24 09:29:17.000
+瑞泰,ef4254c21e6547e092c033780c758916,西安瑞泰口腔医院太白分院,,242,61,0,46,133,2,9,0,52,5,0,2026-10-05 11:34:05.000
+瑞泰,5b4a90f4f19f484691e2aec68e88b417,重庆瑞泰口腔汽博门诊,,489,60,0,134,294,1,12,0,49,10,0,2026-10-12 15:00:43.000
+瑞泰,c1021ae5c2334777ac7920a8518562f6,成都瑞泰口腔医院银泰分院,,324,58,0,78,188,0,11,0,47,12,0,2026-10-15 17:46:46.000
+瑞泰,e1ce24a05f7a48029233821cf231b88c,江苏瑞泰口腔江阴美嘉欣门诊,,372,58,0,141,173,0,8,0,51,7,0,2026-09-11 10:08:00.000
+瑞泰,70525b35ab1e408a9a0da5e7e451b005,长沙瑞泰科尔雅口腔医院西奥分院,,210,57,0,65,87,1,7,0,50,2,0,2025-10-07 00:15:18.000
+瑞泰,859e5c7401a5483697a5a24e6d1cf103,北京瑞泰口腔医院陶然亭分院,,371,57,0,50,264,0,3,0,56,4,0,2026-09-14 14:00:39.000
+瑞泰,dcbfb4173cbd41bc94e964cabef92567,济南瑞泰济东口腔医院,,221,56,0,26,127,12,6,0,50,1,0,2023-08-29 10:30:29.000
+瑞泰,99a60449410d4f05af232da8560bda93,成都瑞泰口腔医院建设分院,,250,55,1,47,148,0,11,0,44,4,1,2026-01-23 11:13:50.000
+瑞泰,23a773fa1f464d05aa0aa00e87e4cddb,重庆瑞泰口腔巴南门诊,,311,53,0,108,142,8,12,0,41,3,0,2026-10-13 10:01:19.000
+瑞泰,d673c106bb614ef19130c24fc5450581,长沙瑞泰科尔雅口腔医院齐美分院,,195,52,0,38,102,3,10,0,43,1,0,2026-10-14 11:01:38.000
+瑞泰,69165bd994ec49429dcc66cf8c3c54d6,重庆瑞泰口腔大坪门诊,,383,49,0,242,92,0,11,0,38,5,0,2025-08-16 16:00:32.000
+瑞泰,1baf80fbdd7842a19e63b19e46d90d9b,西安瑞泰口腔医院高科分院,,268,49,1,108,110,1,5,0,44,5,0,2026-10-17 16:42:34.000
+瑞泰,6cd7d4408e7e46ad9c9fad0b764e6d66,重庆瑞泰口腔大学城门诊,,791,49,3,329,361,52,14,0,35,13,2,2026-10-12 12:18:38.000
+瑞泰,46daed76087f4aa9a36709413c21d9e7,西安瑞泰口腔医院龙首分院,,337,47,1,77,212,1,11,0,36,2,0,2026-10-01 16:07:19.000
+瑞泰,ce6db69026b44347b527b0158a05bb8f,北京瑞泰口腔医院亦庄分院,,250,47,1,62,140,1,1,0,46,7,0,2026-10-16 19:04:53.000
+瑞泰,e83d432a38bb4f6284713b36db4e7497,上海瑞泰口腔世纪公园门诊,上海瑞泰口腔世纪公园门诊,245,44,3,31,156,14,10,0,34,34,1,2026-09-01 16:10:45.000
+瑞泰,830e216eed28413cb57c00beb71a1854,北京瑞泰口腔医院上地分院,,274,44,0,18,212,0,0,0,44,0,0,2025-08-12 17:41:47.000
+瑞泰,358c9c024e254d3586856aab770eb217,重庆瑞泰华西口腔巴国城门诊,,428,41,0,223,163,1,7,0,35,4,0,2025-06-21 11:46:22.000
+瑞泰,83f93882fa884b96bb3bb9fdddb75541,重庆瑞泰久悦口腔沙坪坝门诊,,697,41,0,228,428,0,9,0,32,8,0,2025-06-29 17:01:22.000
+瑞泰,c50b5e509354470889020d5674cb888f,成都瑞泰口腔医院世纪城分院,,219,40,0,48,126,5,19,0,25,5,0,2026-10-17 15:20:47.000
+瑞泰,81d67cc568f441b78a0da3ee9dffe96b,青岛瑞泰口腔江西路门诊,,345,39,0,77,228,1,2,0,37,1,0,2026-10-18 08:37:34.000
+瑞泰,f338d89191d64056964966cc9c5d23a2,长沙瑞泰科尔雅口腔医院北辰分院,,242,39,0,86,116,1,3,0,36,1,0,2025-09-03 16:03:21.000
+瑞泰,d0d5976bf8924544a59fd6cb0bed3b96,北京瑞泰双井分院,,186,36,1,36,114,0,1,0,35,1,0,2025-01-10 14:33:58.000
+瑞泰,ce07474115f5463586605381049cb45d,重庆瑞泰久悦口腔源著门诊,,328,36,0,138,154,0,6,0,30,3,0,2024-07-12 17:50:53.000
+瑞泰,3a9ee869eb4e4121bcafe6fef490f9f6,重庆瑞泰口腔英利门诊,,233,35,1,86,111,1,10,0,26,1,0,2026-10-16 12:00:59.000
+瑞泰,b44d5e738fb842278e69a29bd76929ad,靖江瑞泰口腔门诊部,,124,35,0,27,62,0,5,0,31,3,0,2023-05-05 09:57:59.000
+瑞泰,384741cdca30437e81ff4477ddc30979,重庆瑞泰口腔长寿门诊,,355,35,0,117,178,25,10,0,25,4,0,2026-10-12 16:52:39.000
+瑞泰,e147b68f65a14c33a80299d34040f92d,青岛瑞泰东和齿科同安路门诊,,142,33,0,27,82,0,14,0,21,3,0,2024-09-14 18:01:01.000
+瑞泰,30f4f6702e22441a9ac40e1141fe739a,长沙瑞泰科尔雅口腔医院麓南分院,,136,31,0,22,81,2,10,0,22,1,0,2025-06-05 15:33:19.000
+瑞泰,f07ae73a506d4aaa99c76091fb2199f8,北京瑞泰广外分院,,290,29,0,83,59,119,1,0,29,3,0,2026-09-11 13:30:34.000
+瑞泰,829fe22dc2c044fc962b4cd981b88af8,重庆瑞泰口腔礼嘉门诊,,261,29,0,111,121,0,8,0,22,3,0,2024-09-10 15:28:07.000
+瑞泰,924a6a4489f74f43886e8e07c679fa5f,西安瑞泰口腔医院曲江分院,,305,26,0,95,184,0,1,0,25,4,0,2026-07-22 11:39:46.000
+瑞泰,9100447bec654e689e4f7f42b202d361,成都瑞泰口腔医院悦街分院,,277,22,2,84,171,0,10,0,12,2,0,2026-06-27 16:23:54.000
+瑞泰,aca95973f9ac43d0b5cee4cb25f7330e,青岛瑞泰口腔五四广场门诊,,124,22,0,16,86,0,7,0,15,0,0,2024-09-02 09:07:38.000
+瑞泰,c78931a84e244bf99bcba68e4dc02a66,成都瑞泰口腔医院西宸分院,,160,19,0,57,84,0,10,0,9,2,0,2024-08-12 14:54:01.000
+瑞泰,4e86b5a2a23e4f2baa541643d541d152,重庆瑞泰口腔U城天街门诊,,479,18,3,295,147,19,10,0,8,0,0,2026-07-03 15:02:30.000
+瑞泰,d2b64687bc4a41418f0cbc5f03b0a44f,长沙瑞泰科尔雅口腔医院昊天分院,,196,13,0,46,137,0,3,0,10,2,0,2025-07-24 17:26:22.000
+瑞泰,b905a2bc04ef4abfa3bcc47f3d695e36,大连瑞泰口腔和平广场门诊,,77,12,1,20,38,7,4,0,8,1,0,2026-09-16 16:14:56.000
+瑞泰,f3d67e38ffd847128590566bded9bd1e,大连瑞泰口腔三八广场门诊,,93,11,1,39,43,0,5,0,6,0,0,2025-08-19 14:19:32.000
+瑞泰,c686add3b0254eadbcb27d07e9702b90,上海瑞泰静安昌平路门诊,,228,7,1,44,154,23,4,0,3,0,0,2024-11-27 21:22:51.000
+瑞泰,48cd2a167556491c8bfe65b19c86cc26,北京瑞泰云岗分院,,75,6,0,4,65,0,0,0,6,0,0,2025-06-13 14:45:17.000

docs/qa/2026-10-18-jvs-complaint-ingestion-local-validation.md

+# 5i5ya 投诉标准化与增量采集：本地验收
+
+日期：2026-10-18（Asia/Shanghai）
+
+分支：`codex/jvs-complaint-ingestion`
+
+基线：`main` / `76e7d5bc69e98d9f61d4a967c131c4a58d3d0207`
+
+**投诉摄入已实现并完成本地验证。未提交、未合并、未部署；没有修改召回规则、画像提取器或免打扰字段。**
+
+## 1. 接入方式
+
+沿用现有链路：
+
+`5i5ya 数仓 → jvs-dw manifest → transform / assembler → transaction → parser → patient_facts`
+
+- 源表为 `dw_group.fact_complain_out`，没有绕过现有摄入另建定时任务。
+- 患者沿用 `tenant_id=ruier-grp` 和 `identity_namespace_field=brand`，按宿主、租户、命名空间、患者外部 ID 关联。
+- 反馈单按配置的 `(brand,id)` 聚合；投诉编号只用于展示。通用聚合算子不硬编码品牌或 5i5ya 字段。
+- `clinicId` 保留投诉单的 `organization_id`，不改成患者最近就诊门店，不因名称字典缺失过滤诊所。
+- 共用现有事务账本和事实版本表，无新数据库表、迁移或后台服务。
+
+## 2. 标准化口径
+
+标准字段包括：单 ID、患者 ID、诊所、发生时间、版本时间、反馈性质、投诉类别、渠道、处理状态、正文、结案时间及完整跟进数组。
+
+| 信息 | 实现口径 |
+|---|---|
+| 性质 | 关注点明确为负面才标 `complaint`；表扬、中性、未知分别保留，不靠评分猜投诉 |
+| 分类冲突 | 明确的服务/医疗投诉分类与关注点性质冲突，标未知并保留原因；源默认“其他投诉”不覆盖关注点 |
+| 状态 | 未处理→`open`、处理中→`in_progress`、已完成→`resolved`、草稿→`draft`；未知不默认未处理 |
+| 有效信号 | 性质为投诉，且状态为未处理、处理中或已完成；严重度没有可靠来源，留空 |
+| 非投诉快照 | 共用版本链，以 `invalidated` 和 `signal_eligible=false` 保存；用于撤销先前投诉和阻止旧版本恢复，不计为有效投诉 |
+| 跟进 | 一单多行还原为完整数组；相同跟进去重、稳定排序，保留动作原文和人员来源信息 |
+| 结案时间 | 已完成时取“完成”跟进时间；没有则空，不用采集时间冒充 |
+| 时间 | 主单、创建、源更新、跟进均按 `Asia/Shanghai` 解释；事实/事务落库为 UTC |
+| 原文 | 保留完整原始多行封套，可从账本离线重新标准化，不依赖再次查询源端 |
+
+`updatedAt` 是本次完整反馈快照的版本时间：`max(created_gmt_at, updated_gmt_at, follow_date)`；源主单原更新时间另存 `sourceUpdatedAt`。因此只有跟进更新，也能正确升级快照。
+
+缺患者、缺诊所、非法时间、同键主体冲突不能强行归属；标准化阶段按单拒绝并接入现有拒绝举证。正常增量发现会先排除无患者编号的源行，这部分无法进入患者 cohort，需要保留源侧核查统计，不能声称它们已进入 PAC 拒绝账本。
+
+## 3. 增量与幂等
+
+**先找变化的患者，再读取该患者完整的投诉及跟进。**
+
+- `complaint_changes` 查询用上述三个时间的最大值作游标；SQL 只过滤无法关联的患者身份，不过滤成“仅未完成”或“仅投诉”，以便捕获结案和改分类。
+- `fact_complain_out` 完整查询只加现有患者 cohort 范围，不再加时间窗，避免老跟进丢失。
+- 新资源没有历史游标时，通过显式 `bootstrap_full` 发现存量患者。该开关只启用于投诉变化查询，不改变其他资源的默认行为。
+- 继续使用既有调度、游标、48 小时回看和患者分批机制。首轮可能带入较多患者；这些患者的其他资源仍按现有 cohort 管线处理，不是另开全宿主重解析或重算任务。
+- 内容变化参与事务幂等键；同源快照乱序、重复输入保持同一键。事实使用已有快照水位/事件序号防护，同时间的分类修订能够替代旧状态，旧报文重试不能恢复旧投诉。
+- 空转探针复用完整增量 SQL，支持 `changed_at` 派生别名，避免把它误当作物理列。
+
+## 4. 实测结果
+
+### 真实源样本 → 本地 PostgreSQL
+
+只读抽取 **139 行、79 张反馈单**；样本含不同品牌、性质、状态、跨品牌同号、空更新时间、跟进晚于主单、缺患者等情况。原始样本仅保存在本机私有临时目录，不加入仓库。
+
+| 检查 | 结果 |
+|---|---|
+| 可标准化并落库 | 65 张 |
+| 缺患者编号拒绝 | 14 张 |
+| 有效投诉 | 21 张：5 张未完成、16 张已完成 |
+| 无效反馈快照 | 44 张：非投诉、未知、草稿或冲突，均无有效投诉信号 |
+| 全样本再次摄入 | 事务、事实均无重复新增 |
+| 投诉改为“无不满”，不修改源时间 | 旧投诉效力撤销；旧报文重放不复活 |
+| 从已存原文批量重放历史版本 | 无新增或错误覆盖，当前版本保持正确 |
+| 同号跨品牌 | 分别落库，不互相去重或覆盖 |
+| 同号改挂另一个患者 | 拒绝，不搬移事实 |
+| 北京跨日时间 | `2026-10-17 00:15` → `2026-10-16T16:15:00Z`；所有样本事实发生/版本时间与规范化结果逐一一致 |
+
+验证使用独立本地数据库 `pac_complaint_verify_20261018`，应用现有迁移；测试完成后清除本次测试宿主及样本。未触碰本地应用数据库的数据，未启动/重启应用或容器。
+
+### 生成 SQL → 数仓只读验证
+
+直接执行新采集器生成的查询，限制单线程和查询时限：
+
+- 首轮存量发现：**19,114 位患者**（品牌＋患者 ID 去重；不是投诉人数）。
+- 选取存在“跟进晚于主单更新时间”的真实患者：增量查询 **2 行**，完整查询 **4 行**。
+- 确认新跟进被发现，窗口前的旧跟进仍保留。
+- 派生游标探针成功执行，该验证窗口返回 **2,386 行**；不是本轮实际导入数。
+
+### 自动验证
+
+- 相关回归：**17 组、345 项通过**；另 1 项为可选 SQL 导出步骤，已单独启用运行并用于上述实测。
+- 覆盖投诉标准化、身份隔离、时区、完整聚合、重复/乱序重放、事实批写，以及原有病历、影像、cohort、原文追溯和增量探针。
+- 类型检查：后端源码及 tests 一起检查通过；共享类型构建通过。
+- 没有运行召回全量计算。
+
+## 5. 仍然存在的源端边界
+
+1. 数仓没有 ETL 入仓时间和删除标记。48 小时回看能覆盖窗口内迟到，**不能保证发现窗口外的回填、原地修改但所有时间均未变化、物理删除**。这些需要源端补充变更保证或后续对账。
+2. 跟进没有独立 ID，相同内容只能按规范化内容去重，无法区分两次内容完全一致的真实跟进。
+3. 分类含义不清、患者身份缺失的记录不猜测；目前出口覆盖范围仍以源数据评估为准。
+4. 本次接通数据，不表示已有投诉展示或召回避让效果；投诉如何影响召回另行设计。
+
+源码入口：[manifest](../../apps/pac-service/data/jvs-dw/manifest.yaml)、[投诉 assembler](../../apps/pac-service/data/jvs-dw/assemblers/complaint.yaml)、[标准字段](../../packages/types/src/schemas/canonical.ts)、[parser](../../apps/pac-service/src/modules/sync/pipeline/parsers/complaint.parser.ts)。
+
+初始源数据调查见 [摄入评估](./2026-10-18-jvs-complaint-ingestion-assessment.md) 和 [70 家诊所汇总](./2026-10-18-jvs-complaint-clinics.csv)。初始评估描述的是修改前状态，以本验收说明为当前本地实现状态。

packages/types/src/schemas/canonical.ts

@@ -496,25 +496,47 @@ export const RechargeCanonicalSchema = z
   .passthrough();
 export type RechargeCanonical = z.infer<typeof RechargeCanonicalSchema>;
 
-/// 投诉记录(合规营销避雷的关键信号)
-/// 产品收集
-export const ComplaintCanonicalSchema = z
-  .object({
-    externalId: z.string().min(1),
-    patientExternalId: z.string().min(1),
-    clinicId: z.string().min(1),
-    filedAt: isoDateTime,
-    /// 投诉内容自由文本
-    content: z.string().optional().nullable(),
-    /// 关联治疗(如有)— 引用 treatment_record subject_id 或 host 治疗 id
-    relatedTreatmentSubjectId: z.string().optional().nullable(),
-    /// 投诉类别(host 自定义,如 service / clinical / billing)
-    category: z.string().optional().nullable(),
-    /// 处理状态(open / resolved / pending)— host 编码,assembler.yaml enum_mapping 翻译
-    status: z.string().optional().default('open'),
-    resolvedAt: optionalIsoDateTime,
-  })
-  .passthrough();
+/// 投诉/反馈快照：是否形成有效投诉信号与源记录留存分开。
+export const ComplaintStatusSchema = z.enum(['open', 'in_progress', 'resolved', 'draft', 'unknown']);
+export const ComplaintFeedbackKindSchema = z.enum(['complaint', 'praise', 'neutral', 'unknown']);
+export const ComplaintCategorySchema = z.enum(['service', 'clinical', 'billing', 'environment', 'waiting', 'other', 'unknown']);
+export const ComplaintChannelSchema = z.enum(['sms_survey', 'hotline', 'third_party', 'sales', 'mini_program', 'unknown']);
+const complaintDateTime = z.iso.datetime({ offset: true });
+const optionalComplaintDateTime = z.preprocess(v => v === '' ? null : v, complaintDateTime.nullable().optional());
+export const ComplaintFollowUpSchema = z.object({
+  occurredAt: optionalComplaintDateTime,
+  action: z.enum(['transferred', 'completed', 'edited', 'followed_up', 'paused', 'unknown']),
+  sourceAction: z.string().nullable().optional(),
+  content: z.string().nullable().optional(),
+  actorId: z.coerce.string().nullable().optional(),
+  actorName: z.string().nullable().optional(),
+});
+const complaintId = z.union([z.string().trim().min(1), z.number().int().safe()]).transform(String);
+export const ComplaintCanonicalSchema = z.object({
+  externalId: complaintId,
+  patientExternalId: complaintId,
+  clinicId: complaintId,
+  filedAt: complaintDateTime,
+  createdAt: optionalComplaintDateTime,
+  updatedAt: optionalComplaintDateTime,
+  sourceUpdatedAt: optionalComplaintDateTime,
+  content: z.string().optional().nullable(),
+  relatedTreatmentSubjectId: z.string().optional().nullable(),
+  relatedOrderExternalId: z.string().optional().nullable(),
+  number: z.string().optional().nullable(),
+  feedbackKind: z.preprocess(v => v == null ? undefined : v, ComplaintFeedbackKindSchema.default('unknown')),
+  // 宿主另一分类维度，仅用于识别互相矛盾的证据；没有则未知。
+  classificationKind: z.preprocess(v => v == null ? undefined : v, ComplaintFeedbackKindSchema.default('unknown')),
+  category: z.preprocess(v => v == null ? undefined : v, ComplaintCategorySchema.default('unknown')),
+  channel: z.preprocess(v => v == null ? undefined : v, ComplaintChannelSchema.default('unknown')),
+  status: z.preprocess(v => v == null ? undefined : v, ComplaintStatusSchema.default('unknown')),
+  sourceStatus: z.string().optional().nullable(),
+  sourceCategory: z.string().optional().nullable(),
+  sourceConcern: z.string().optional().nullable(),
+  sourceChannel: z.string().optional().nullable(),
+  resolvedAt: optionalComplaintDateTime,
+  followUps: z.array(ComplaintFollowUpSchema).default([]),
+}).passthrough();
 export type ComplaintCanonical = z.infer<typeof ComplaintCanonicalSchema>;
 
 /// 转介绍记录(老客带新)
@@ -595,6 +617,7 @@ export const CanonicalResourceMeta: Record<
     moneyFields: readonly string[];
     moneyArrayFields: readonly { array: string; field: string }[]; // 数组里嵌套 money 字段
     datetimeFields: readonly string[];
+    datetimeArrayFields?: readonly { array: string; field: string }[];
     booleanFields: readonly string[];
   }
 > = {
@@ -716,7 +739,8 @@ export const CanonicalResourceMeta: Record<
   complaint: {
     moneyFields: [],
     moneyArrayFields: [],
-    datetimeFields: ['filedAt', 'resolvedAt'],
+    datetimeFields: ['filedAt', 'resolvedAt', 'sourceUpdatedAt'],
+    datetimeArrayFields: [{ array: 'followUps', field: 'occurredAt' }],
     booleanFields: [],
   },
   referral: {