# JVS DW Cold Import Manifest(v2.3 — 单 host 多 tenant per-row 路由)
#
# 数据源:阿里云 ClickHouse(集团 DW ADS 出口层)
# 试点 5 诊所(双 brand 同一 manifest 跑完):
#   瑞尔(tenant=ruier)
#     7d49539c7573490387c03e6496ff1a6c  杭州大厦诊所
#     dad2f04a120947e2b82b41cbd108f3f4  杭州高德诊所
#     66701845dd2342e19f9e9f576c4ffe9c  北京朝阳公园诊所
#   瑞泰(tenant=ruitai)
#     c18cadf2d3cd4adda5527debd41356eb  通善口腔学前街医院
#     e83d432a38bb4f6284713b36db4e7497  上海世纪公园
#
# ─────────────────────────────────────────────────────────────
# 架构纪律(v2.2 — yaml 单方案改造,docs/algorithm/canonical-fact-layer.md §五)
# ─────────────────────────────────────────────────────────────
#
# 数据源(双源完全等价):
#   - 直连模式:sql_source.queries 每条 SQL **只做"等价 csv 单表 dump"的 SELECT**
#     即:列选 + WHERE 过滤(cohort 限定),其他 host 端可在 dump 时做同样过滤产出 csv
#   - csv 模式(W6+):host 端用同 WHERE 条件 dump 出来,PAC 切 tables[] 配置即可,
#     transforms / assemblers 一行不动
#
# **SQL 禁止做**(违反双源等价 + 违反 canonical-fact-layer §一 原则 1):
#   ❌ ARRAY JOIN(JSON 数组拆行)— 改 transforms.split_json_array
#   ❌ JSONExtract*(嵌套字段提取)— 同上,split_json_array.element_fields
#   ❌ substring / 字符串变换 — 改 transforms.derive
#   ❌ multiIf / CASE WHEN 关键词分流 — 改 yaml enum_mapping 或 transforms.route_by_pattern
#   ❌ multiIf pay_xxx > 0 推单列 — 改 transforms.pick_first_nonzero
#   ❌ 复杂 GROUP BY / 跨表 JOIN — 改 AssemblerEngine join_arrays 或 transforms
#
# **SQL 允许**:
#   ✅ SELECT col1, col2 [AS alias] FROM single_table
#   ✅ WHERE col=... AND col IN (...)
#   ✅ patient_id IN (SELECT patient_id FROM ... LIMIT n)  ← cohort 限定(csv 模式 host 跑同样 SQL)
#
# 跑法:
#   export DW_CLICKHOUSE_PASSWORD='<密码>'
#   pnpm cold-import -- --dir=./data/jvs-dw --dry-run
#   pnpm cold-import -- --dir=./data/jvs-dw

host_name:     jvs-dw
display_name:  瑞尔集团 DW(单 host 多 brand)
auto_sync:     true   # 纳入每日自动增量同步(scheduler 自动发现;取代 PAC_INCREMENTAL_HOSTS env)

# ── tenant 路由(临时方案,W3 末)──
# 理想态:DW 所有 ADS 表都带 tenant_id UUID 字段,PAC 直接 tenant_field=tenant_id pass-through。
# 现状:DW 只有 fact_emr_treatment_out 一张表带 tenant_id,其他表只有 brand 中文字段。
#   - 瑞尔 ↔ ba67e6cf30dc4f9c9c46adef188bbd04(DW 真实 host tenant UUID)
#   - 瑞泰 ↔ 77057aed269f4a14957ae0ad0eff359a
# 临时:yaml 维护 brand → DW UUID 映射,PAC 内部用 UUID 作 tenant_id(英文 / 跨表稳定)。
# 后续:已反馈 DW 团队补全所有表的 tenant_id 字段;补完后改 tenant_field=tenant_id 删本映射。
tenant_field:  brand
tenant_map:
  瑞尔: ba67e6cf30dc4f9c9c46adef188bbd04
  瑞泰: 77057aed269f4a14957ae0ad0eff359a

amount_unit:   yuan
timezone:      Asia/Shanghai

# ─────────────────────────────────────────────────────────────
# 数据源(SQL 朴素化:只 SELECT + WHERE)
# ─────────────────────────────────────────────────────────────
sql_source:
  kind: clickhouse
  connection:
    # 本地 DW 镜像(docker-compose clickhouse,全量复制自瑞尔 DW)。
    # 切回远程 DW:url 换成 http://cc-2zen8w29e49076o83.public.clickhouse.ads.aliyuncs.com:8123
    #             username: jvs_pac,password_env 设 DW 真实密码。
    url: http://localhost:8123
    database: dw_group
    username: default
    password_env: DW_CLICKHOUSE_PASSWORD
  # 1000 cohort 患者 settlement 总行 ~ 20 万(p50=12/患者,max=11.6 万/异常患者)
  # 100k 会截一半数据,关键事实(吴建康 61 行种植结算)被丢失 → scenario 错召回
  # 全量(13 万患者)dry-run 实测:appointment / diagnosis 单表 ~100-130 万行
  # → 200 万兜底(Node heap 8GB 够),长期(生产)走增量 pull(cursor)替代一次性拉
  default_limit: 2000000

  # W4 末:DW 增量配置(sync-incremental CLI 用)
  # 跑全量(cold-import)时本段忽略;跑增量时按 per_query 注入 WHERE
  # 首次增量(无 cursor)= 等价全量(跟 cold-import 不同点 = 增量模式去 dev cohort LIMIT,不限量)
  # fact_client_out 没 updated_date,用 last_visit_time(主档变化 = 患者来诊新事件)
  incremental:
    per_query:
      fact_client_out:           { cursor_column: last_visit_time }
      fact_appointment_out:      { cursor_column: updated_date }
      fact_emr_treatment_out:    { cursor_column: updated_date }
      fact_settlement_out:       { cursor_column: updated_date }
      fact_settlement_out_refund: { cursor_column: updated_date }
      fact_settlement_mode_out:  { cursor_column: updated_date }

  # ⭐ Cohort 分批配置(PR2/PR4)— 宿主无关性的关键
  # PAC 摄入代码不硬编码任何表名/列名,全从这里读;别的 host 接入只改本段
  cohort:
    patient_list_from:   dw_group.fact_client_out  # 列患者清单的源表(患者主档表)
    patient_key_column:  patient_id                # 患者主键列(所有源表共用此列名做 cohort 过滤)
    tenant_key_column:   brand                     # 租户区分列(jvs-dw 多 brand;单 tenant host 删本行)
    list_cursor_column:  last_visit_time           # 列患者时增量 cursor 列(对应患者主档表时间列)

  # SQL 最朴素化 — host(DW)给的数据范围就是 PAC 该消化的范围。
  # PAC 这边不预设诊所 / brand / 时间过滤,数据来什么就是什么。
  # dev 期的 cohort LIMIT 已去除 → 全量;只保留必要业务过滤(退费状态)。
  # cohort 用 (patient_id, brand) 复合键 — patient_id 跨 brand 撞车,需带 brand 才精确。
  queries:
    # ── 患者主档 ──
    fact_client_out: |
      SELECT *
      FROM dw_group.fact_client_out
      WHERE last_visit_time IS NOT NULL

    # ── 预约(全部状态;in_time NULL 的在 transforms.route 时分流跳过 encounter) ──
    fact_appointment_out: |
      SELECT *
      FROM dw_group.fact_appointment_out
      WHERE (patient_id, brand) IN (
        SELECT patient_id, brand FROM dw_group.fact_client_out
        WHERE last_visit_time IS NOT NULL
      )

    # ── EMR 全字段(自由文本 + diag/treat_plan JSON,后续 transforms 拆) ──
    fact_emr_treatment_out: |
      SELECT
        id, patient_id, organization_id, brand, rq, registration_id, user_id, doctor_name,
        illness_desc, pre_illness, past_hist, gen_cond, examine, dispose, doc_order,
        diag, treat_plan, plan, file_url,
        created_date, updated_date
      FROM dw_group.fact_emr_treatment_out
      WHERE (patient_id, brand) IN (
        SELECT patient_id, brand FROM dw_group.fact_client_out
        WHERE last_visit_time IS NOT NULL
      )

    # ── 结算明细(已做治疗反推;退费/未完成结算行 PAC 不消费) ──
    # created_date(结算记录入库时间)比 billing_date(开单/收费动作)更接近治疗实际发生:
    #   - 100% created_date >= billing_date(永远不会更早)
    #   - 23% case 团购券/储值/挂号预付,billing_date 在治疗前几小时,created_date 才是医生录治疗的时点
    #   - 全量 244 万对验证:settlement.created vs EMR.created p50 差 5min,51% 在 5min 内
    fact_settlement_out: |
      SELECT id, organization_id, brand, patient_id, rq, billing_date, created_date, updated_date,
             settlement_type, doctor_id,
             settlement_project_name, settlement_money, settlement_num, settlement_unit_name,
             registration_id, is_refund, settlement_status
      FROM dw_group.fact_settlement_out
      WHERE (is_refund IS NULL OR is_refund = 0)
        AND settlement_status = 1
        AND (patient_id, brand) IN (
          SELECT patient_id, brand FROM dw_group.fact_client_out
          WHERE last_visit_time IS NOT NULL
        )

    # ── 退费明细(独立 fact_type refund_record;LTV 算法 - refund;S5 闭环风控)──
    # 双轨判定 — host 退费两种表达方式都收:
    #   ① is_refund=1            (显式标记的退费单,settlement_money 通常 > 0 表示"被退金额")
    #   ② settlement_status=4    (反向结算单,settlement_money 通常 < 0 表示"金额冲减")
    # parser 侧 Math.abs(amount) 统一成正 cents(语义 = 患者拿回的钱)
    fact_settlement_out_refund: |
      SELECT id, organization_id, brand, patient_id, rq, billing_date, created_date, updated_date,
             settlement_type, doctor_id,
             settlement_project_name, settlement_money,
             registration_id, is_refund, settlement_status
      FROM dw_group.fact_settlement_out
      WHERE (is_refund = 1 OR settlement_status = 4)
        AND (patient_id, brand) IN (
          SELECT patient_id, brand FROM dw_group.fact_client_out
          WHERE last_visit_time IS NOT NULL
        )

    # ── 支付通道明细(17 列 pay_*,后续 transforms.pick_first_nonzero 推) ──
    # WHERE settlement_status='1' 过滤掉退费行(status='4' 金额负数,该归 refund_record)
    fact_settlement_mode_out: |
      SELECT *
      FROM dw_group.fact_settlement_mode_out
      WHERE settlement_status = '1'
        AND (patient_id, brand) IN (
          SELECT patient_id, brand FROM dw_group.fact_client_out
          WHERE last_visit_time IS NOT NULL
        )

    # ── 影像 AI 分析(fact_emr_image_analysis_out)→ 诊断信号源(image_finding)──
    # 结构化 AI 源,源 SQL 一次 pivot+炸牙位(下游零 transform):
    #   join file_num→client 取 patient_id+brand;org 用患者 EMR organization_id 作诊所(image 无诊所);
    #   ARRAY JOIN 把 10 病种列 pivot 成 (code,牙位数组),splitByChar+arrayJoin 炸成每牙一行。
    # ⚠️ prod CH 23.8 不允许「多 JOIN + ARRAY JOIN 同层」→ 必须分层嵌套:
    #   s1(joins,无 array join)→ s2(array join,读 s1 子查询)→ 外层(无 join)。
    # ⚠️ cohort 注入(裸 patient_id,brand IN)会插到首个 GROUP BY 前 = po 子查询 → po 按 batch scope,
    #   外层 notEmpty(po.org) 恰好把输出过滤到 batch 患者(已在 prod CH 验证等价正确)。
    # 病种→K 码映射留 manifest(host 形态);去重靠召回 (subKey,tooth) 聚类;code_source=image_ai 独立 subject。
    image_finding_rows: |
      SELECT patient_id, brand, organization_id, emr_id, rq, code, code_source, tooth,
             concat(emr_id, '|imgai|', code, '|', tooth) AS diag_external_id
      FROM (
        SELECT patient_id, brand, organization_id, emr_id, rq, cm.1 AS code, 'image_ai' AS code_source,
               -- ⭐ 牙位整组保留一行(用 ; 拼接),对齐医生诊断粒度(一个诊断码一条 fact,非逐牙拆行)。
               -- '['15','25']' → '15;25';召回侧 union-find 仍按牙逐颗扣除/聚类。
               -- 注:同 K 码跨影像列(K01 阻生+埋伏 / K03 三列)仍各列一条 fact,union-find 会重聚类。
               replaceRegexpAll(replaceRegexpAll(cm.2, '[\[\] '']', ''), ',', ';') AS tooth
        FROM (
          SELECT c.patient_id AS patient_id, c.brand AS brand, po.org AS organization_id,
                 ia.emr_id AS emr_id, ia.rq AS rq,
                 ia.cavity AS cavity, ia.impacted_tooth AS impacted_tooth, ia.embedded_tooth AS embedded_tooth,
                 ia.root_periodontitis AS root_periodontitis, ia.root_remnant AS root_remnant,
                 ia.crown_remnant AS crown_remnant, ia.wedge_shaped_defect AS wedge_shaped_defect,
                 ia.cyst AS cyst, ia.tooth_loss AS tooth_loss, ia.retained_primary_tooth AS retained_primary_tooth
          FROM dw_group.fact_emr_image_analysis_out ia
          INNER JOIN dw_group.fact_client_out c ON c.file_num = ia.file_num AND c.brand = ia.brand
          LEFT JOIN (
            SELECT patient_id, brand, any(organization_id) AS org
            FROM dw_group.fact_emr_treatment_out WHERE notEmpty(organization_id)
            GROUP BY patient_id, brand
          ) po ON po.patient_id = c.patient_id AND po.brand = c.brand
          WHERE c.last_visit_time IS NOT NULL AND notEmpty(po.org)
        ) s1
        ARRAY JOIN [('K02', cavity), ('K01', impacted_tooth), ('K01', embedded_tooth),
                    ('K04', root_periodontitis), ('K03', root_remnant), ('K03', crown_remnant),
                    ('K03', wedge_shaped_defect), ('K09', cyst), ('K08', tooth_loss),
                    ('K00', retained_primary_tooth)] AS cm
        WHERE cm.2 != '[]' AND cm.2 != ''
      ) s2
      WHERE tooth != ''

    # ── 诊所回访任务(fact_returnvisit_out)→ patient_return_visit upsert(展示用,5 试点)──
    # customer_id AS patient_id(让 cohort 过滤生效);WHERE org ∈ 5 试点(= EMR 表的 org)→ 只摄 5 家。
    # 非召回信号,详情页"回访记录"块展示(常规/术后/咨询回访)。
    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
      FROM dw_group.fact_returnvisit_out
      WHERE organization_id IN (SELECT DISTINCT organization_id FROM dw_group.fact_emr_treatment_out)

    # ── 咨询主体(fact_consult_out)→ consultation_record(意向源,5 试点)──
    # patient_register_id = patient_id(5 诊所内 94.4% 命中;CH 允许 WHERE 引用别名,同 returnvisit)。
    # SQL 朴素(只 SELECT 列 + WHERE 过滤,CSV 双源等价);consult_external_id 由 transforms.derive 拼(I 段)。
    #   无 updated_date/无 id → external_id=(patient,appo,date);不入 per_query(无 cursor)→ 每轮全量按
    #   org+patient 过滤再拉(幂等 upsert,同 returnvisit)。potential_cure=患者意向(主观,非诊断,不进召回)。
    fact_consult_out: |
      SELECT patient_register_id AS patient_id, organization_id, brand,
             appointment_date, first_visit, task_director, potential_cure, doctor_user_id, appo_id
      FROM dw_group.fact_consult_out
      WHERE organization_id IN (SELECT DISTINCT organization_id FROM dw_group.fact_emr_treatment_out)
        AND (patient_register_id, brand) IN (
          SELECT patient_id, brand FROM dw_group.fact_client_out WHERE last_visit_time IS NOT NULL
        )

    # ── 客户-推荐人关系(联系人/亲属边)→ patient_relation upsert ──
    # 一行 = 本人 与 一个关系人(referee)的关系;referee_patient_id=0 = 无关系人,剔除。
    # 本人(patient_id)限 active client(= PAC patient 才能挂靠);关系人不限(解析得到则填 link)。
    fact_customer_referee_out: |
      SELECT patient_id, referee_patient_id, referee_relationship, referee_relationship_name, brand
      FROM dw_group.fact_customer_referee_out
      WHERE referee_patient_id != 0
        AND (patient_id, brand) IN (
          SELECT patient_id, brand FROM dw_group.fact_client_out
          WHERE last_visit_time IS NOT NULL
        )

# ─────────────────────────────────────────────────────────────
# Layer A.5 transforms(SQL 干过的事搬这里,yaml 声明式)
# ─────────────────────────────────────────────────────────────
transforms:
  # ── A. EMR 主表 → 只留自由文本(剥离 JSON 字段) ──
  - kind: project
    input: fact_emr_treatment_out
    output: emr_rows
    keep:
      - id
      - patient_id
      - organization_id
      - brand
      - rq
      - registration_id
      - user_id
      - doctor_name
      - illness_desc
      - pre_illness
      - past_hist
      - gen_cond
      - examine
      - dispose
      - doc_order
      - created_date  # W4 末:DW upsert,emr.yaml field_mapping 拿 updatedAt → source_event_id 幂等键
      - updated_date

  # ── B. EMR.diag JSON 拆行 → 诊断 fact 候选 ──
  # W3 末改:不再丢 stdCode 缺失行(实测 98% 诊断没填 ICD 码,但 message 是标准中文术语)。
  # 全部保留:有 stdCode → 截短取 K 码;无 stdCode → message 归一后交 enum_mapping 映射;
  # 都映射不出的 → 仍落 fact 但 code=null(留 message 给 Layer C LLM 后续补码)。
  - kind: split_json_array
    input: fact_emr_treatment_out
    output: _diagnosis_raw
    array_field: diag
    parent_keys:
      emr_id: id
      patient_id: patient_id
      organization_id: organization_id
      brand: brand
      rq: rq
      user_id: user_id           # 诊断医生(继承 emr 父级,医患关系信号)
      doctor_name: doctor_name   # 诊断医生名(快照)
      created_date: created_date  # W4 末:DW upsert,传父级 created/updated_date 给拆出子行
      updated_date: updated_date
    element_fields:
      std_code: stdCode
      message: message
      tooth_position: toothPosition
    where:
      message: { not_empty: true }    # 只要求有诊断名(message);stdCode 可空

  # B.1 归一 message(去首尾空白/标点 + CJK→ASCII 中段标点)+ stdCode 截短取 K 码
  # W4 末:trim → normalize 升级,解决 host 字段中段中文标点("、""，""（）")字典匹配失败
  - kind: derive
    input: _diagnosis_raw
    output: _diagnosis_norm
    fields:
      message_norm:
        op: normalize
        from: message
      # ⚠️ 宿主 ICD 怪码纠偏:中文临床版把 K07.303 指派给"牙体缺损"(WHO K07=错颌畸形),
      #   naive 截前3位会误成 K07→正畸(全院 1083 患者)。recode 先把 K07.303 改 K03(牙体缺损/修复),
      #   再截短。其余 K07.3xx(牙列不齐/牙拥挤/异位牙)确是正畸,不动。
      std_code_fixed:
        op: recode
        from: std_code
        map:
          K07.303: K03
      std_code_k:
        op: substring
        from: std_code_fixed
        start: 0
        end: 3

  # B.2 coalesce:有 stdCode 用 K 码,否则用归一中文名 → 交 assembler enum_mapping 翻译
  - kind: derive
    input: _diagnosis_norm
    output: diagnosis_rows
    fields:
      diag_code_src:
        op: coalesce
        from: ['std_code_k', 'message_norm']
      diag_external_id:
        op: concat
        parts: ['${emr_id}', '|', '${std_code_k}', '|', '${message_norm}', '|', '${tooth_position}']

  # ── C. EMR.treat_plan(本次治疗 actual)+ EMR.plan(未来计划 planned)双源 ──
  #
  # W4 末重大语义修正(真实数据 + 用户业务确认,详见 dw-data-source-issues.md):
  #   - `treat_plan` 字段 = **本次实际做的治疗(actual)**(字段名误导,真实是 actual)
  #   - `plan` 字段     = **未来计划(planned)**(2023+ host 启用,老数据空)
  #   - 历史上 treat_plan 是杂烩(老数据没 plan 字段,医生把"建议/暂观/复查"也塞这里)
  #     → 现 route_by_pattern 把这些分流走,剩下的才是真治疗 actual,合理
  #
  # 数据流:
  #   treat_plan → split → route:
  #     ① 建议/推荐  → recommendation_rows         (kind=planned,医生当场给的健康路径)
  #     ② 复查/暂观  → treatment_review_rows       (kind=actual,本次做的复查/拆线,chain S4 信号)
  #     ③ 真治疗    → treatment_actual_rows ⭐    (kind=actual,临床真实,带牙位 48.7%)
  #
  #   plan      → split → route:
  #     ① 建议/推荐  → drop(MVP,后续可加 union)
  #     ② 复查/暂观  → drop(避免 kind 冲突)
  #     ③ 真治疗    → treatment_planned_rows ⭐  (kind=planned,医生写好的下一步,带牙位)
  #
  # **settlement → treatment_actual 路径已删除**(原 § D 区段):
  #   - settlement = 财务事件 ≠ 临床事件;一次治疗 5-8 个收费项行 → 反推 actual 颗粒失真
  #   - settlement 0% 带牙位 vs EMR.treat_plan 48.7% 带牙位 → EMR 单源更精确
  #   - settlement 继续走 payment.yaml(LTV)+ refund.yaml(风险),职责单一
  #   - 边界:42% EMR 无 settlement(洁牙赠送/T+延迟)— 用 EMR 单源能正确算 actual
  #   - 反向 0% settlement 无 EMR(实测) — 不会丢治疗

  # ── C.1 split treat_plan(本次治疗) ──
  - kind: split_json_array
    input: fact_emr_treatment_out
    output: _treat_plan_raw
    array_field: treat_plan
    parent_keys:
      emr_id: id
      patient_id: patient_id
      organization_id: organization_id
      brand: brand
      rq: rq
      user_id: user_id           # 治疗医生(继承 emr 父级)
      doctor_name: doctor_name   # 治疗医生名
      created_date: created_date  # W4 末:DW upsert,传父级 created/updated_date 给拆出子行
      updated_date: updated_date
    element_fields:
      treat_name: treatName
      tooth_position: toothPosition
      plan_code: planCode
      treat_stages: treatStages  # W4 末:host 给的 step 数组(开髓/根备/根充等),assembler 翻译到 PAC enum
    where:
      treat_name: { not_empty: true }

  # ── C.2 split plan(未来计划)— W4 末新增 ──
  - kind: split_json_array
    input: fact_emr_treatment_out
    output: _plan_raw
    array_field: plan
    parent_keys:
      emr_id: id
      patient_id: patient_id
      organization_id: organization_id
      brand: brand
      rq: rq
      user_id: user_id           # 制定计划的医生
      doctor_name: doctor_name
      created_date: created_date  # W4 末:DW upsert,传父级 created/updated_date
      updated_date: updated_date
    element_fields:
      treat_name: treatName
      tooth_position: toothPosition
      plan_code: planCode
      treat_stages: treatStages
    where:
      treat_name: { not_empty: true }

  # ── C.2.5 treat_name 归一(头尾空白/标点 + CJK→ASCII 中段标点)──
  # W4 末:host treatName 大量"全口龈上洁治，抛光。" 中文标点变体 + 头空格 + 尾标点
  # 用 normalize op 一次性折叠,下游 route / enum_mapping 字典只需写一种 ASCII 形态
  - kind: derive
    input: _treat_plan_raw
    output: _treat_plan_raw        # in-place 覆盖
    fields:
      treat_name:
        op: normalize
        from: treat_name
  - kind: derive
    input: _plan_raw
    output: _plan_raw              # in-place 覆盖
    fields:
      treat_name:
        op: normalize
        from: treat_name

  # ── C.3 treat_plan 路由分流 ──
  - kind: route_by_pattern
    input: _treat_plan_raw
    field: treat_name
    routes:
      # 建议/推荐(医生当场给的健康路径)→ recommendation_record
      - output: _recommendation_raw
        when:
          starts_with: ['建议', '推荐']
      # 流程性 / 复查 / 行政事项 → treatment_review_rows(本次非治疗事件;chain S4 信号)
      # 注:equals 关键词都是 normalize 后的 ASCII 形态(中标点已折叠);不要在这里写中括号/中逗号
      # &review_terms:YAML anchor — C.9 dispose gate 复用此表(单一真理源,判"treat_plan 无真治疗")
      - output: _treatment_review_raw
        when:
          equals: &review_terms
            # 复查 / 检查 / 复诊
            - 常规复查
            - 复查
            - 定期复查
            - 检查
            - 初诊检查
            - 正畸复查
            - 正畸检查
            - 正畸复诊
            - 种植复查
            - 牙周复查
            - 保持器复诊
            # 暂观 / 观察 / 拆线
            - 拆线
            - 暂观
            - 观察
            - 观察随访
            - 随诊观察
            - 随访观察
            - 无治疗
            - 重复
            - 取资料
            - 暂观,必要时拔除
            - 观察,必要时拔除
            - 随访观察,必要时拔除
            # 流程 / 咨询 / 行政
            - 转诊
            - 请全科医生会诊
            - 正畸咨询
            - 正畸会诊
            - 方案沟通
            - 沟通治疗方案
            - 听方案
            - 缴费
            - 未拍片
            - 拒绝拍片
            - 治疗中
            - 已交付纸质病历
      # 真治疗动作 → treatment_actual_rows ⭐ kind=actual(临床真实)
      - output: _treatment_actual_raw_emr
        when:
          default: true

  # ── C.4 plan 路由分流(plan 的"建议"也接入 recommendation;复查暂 drop)──
  - kind: route_by_pattern
    input: _plan_raw
    field: treat_name
    routes:
      # plan 的"建议"→ recommendation(下面 C.4.5 union 合并到 treat_plan 的 _recommendation_raw)
      - output: _plan_recommendation_raw
        when:
          starts_with: ['建议', '推荐']
      # MVP:plan 的"复查/暂观/流程性/咨询"暂 drop(避免 kind 冲突 + 防止流程性话流入 planned 治疗)
      # future review 信号待后续(transforms 加 union op 后可合并到 treatment_review_rows kind=planned)
      - output: _drop_plan_review
        when:
          equals:
            # 复查 / 检查 / 复诊
            - 常规复查
            - 复查
            - 定期复查
            - 检查
            - 初诊检查
            - 正畸复查
            - 正畸检查
            - 正畸复诊
            - 种植复查
            - 牙周复查
            - 保持器复诊
            # 暂观 / 观察 / 拆线
            - 拆线
            - 暂观
            - 观察
            - 观察随访
            - 随诊观察
            - 随访观察
            - 无治疗
            - 重复
            - 取资料
            - 暂观,必要时拔除
            - 观察,必要时拔除
            - 随访观察,必要时拔除
            # 流程 / 咨询 / 行政
            - 转诊
            - 请全科医生会诊
            - 正畸咨询
            - 正畸会诊
            - 方案沟通
            - 沟通治疗方案
            - 听方案
            - 缴费
            - 未拍片
            - 拒绝拍片
            - 治疗中
            - 已交付纸质病历
      # 真治疗动作 → treatment_planned_rows ⭐ kind=planned(医生写好的下一步)
      - output: _treatment_planned_raw_emr
        when:
          default: true

  # ── C.4.5 合并两源建议 ── treat_plan 的 _recommendation_raw + plan 的 _plan_recommendation_raw
  - kind: union
    inputs: ['_recommendation_raw', '_plan_recommendation_raw']
    output: _recommendation_raw

  # ── C.5 推荐(来自 treat_plan + plan 合并) ──
  - kind: derive
    input: _recommendation_raw
    output: recommendation_rows
    fields:
      rec_external_id:
        op: concat
        parts: ['${emr_id}', '|rec|', '${treat_name}', '|', '${tooth_position}']

  # ── C.6 复查(来自 treat_plan,本次做的复查) ──
  - kind: derive
    input: _treatment_review_raw
    output: treatment_review_rows
    fields:
      treat_external_id:
        op: concat
        parts: ['${emr_id}', '|review|', '${treat_name}', '|', '${tooth_position}']
      # 给所有复查行打上固定 category='review'(yaml enum_mapping 走"_review_sentinel" → review)
      category_raw:
        op: default
        from: __nonexistent_field__
        value: _review_sentinel

  # ── C.7 actual 治疗(来自 treat_plan)⭐ W4 末新增 ──
  - kind: derive
    input: _treatment_actual_raw_emr
    output: treatment_actual_rows
    fields:
      treat_external_id:
        op: concat
        parts: ['${emr_id}', '|tx|act|', '${treat_name}', '|', '${tooth_position}']
      category_raw:
        op: default
        from: treat_name
        value: ''

  # ── C.8 planned 治疗(来自 plan 字段)⭐ W4 末新增 ──
  - kind: derive
    input: _treatment_planned_raw_emr
    output: treatment_planned_rows
    fields:
      treat_external_id:
        op: concat
        parts: ['${emr_id}', '|tx|plan|', '${treat_name}', '|', '${tooth_position}']
      category_raw:
        op: default
        from: treat_name
        value: ''

  # ── C.9 dispose(处置自由文本)→ 补 actual 治疗(仅 treat_plan 无真治疗时)──
  # 背景:处置里常已做治疗(充填/拔除/根管…)。两种关键场景:
  #   ① treat_plan 空 / 仅复查:处置是唯一真治疗源(陈昱天 17:treat_plan=["常规复查"],dispose 树脂充填@17)
  #   ② treat_plan 有真治疗:处置 ~63% 是同次重复叙述(同牙同类)→ 不能再抽(否则双计)
  # gate:treat_plan **空 或 全是复查类**(blank_or_all_in &review_terms)才抽 dispose。
  #   —— 比旧「空才抽」放宽:场景①(treat_plan 仅"常规复查")现在也抽 → 找回被丢的真治疗(解陈昱天误召);
  #      场景②(treat_plan 有真治疗)仍不抽 → 源头不产生重复(fact-writer 终态不去重,必须源头防)。
  # 为何不在 parser 去重:treat_plan/dispose 双记是 host 录入习惯(host 差异)→ 处置归 yaml,parser 保持宿主无关。
  - kind: filter
    input: fact_emr_treatment_out
    output: _emr_dispose_gate
    where:
      treat_plan: { blank_or_all_in: *review_terms }
  - kind: split_json_array
    input: _emr_dispose_gate
    output: _dispose_raw
    array_field: dispose
    parent_keys:
      emr_id: id
      patient_id: patient_id
      organization_id: organization_id
      brand: brand
      rq: rq
      user_id: user_id
      doctor_name: doctor_name
      created_date: created_date
      updated_date: updated_date
    element_fields:
      treat_name: message       # message 当 treat_name:既给 keyword_mapping 分类,又作 subtype
      tooth_position: toothPosition
    where:
      treat_name: { not_empty: true }
  - kind: derive
    input: _dispose_raw
    output: _dispose_tx
    fields:
      treat_external_id:
        op: concat
        parts: ['${emr_id}', '|disp|', '${tooth_position}', '|', '${treat_name}']
      category_raw:           # = message 归一;normalize 折叠 CJK 标点 → keyword strip 切段(建议/必要时)生效
        op: normalize         #   否则 "待观察,若…,建议充填治疗" 全角逗号不切段 → strip 失效 → 误判 restorative
        from: treat_name
  # union 进现有 treatment_actual_rows → 走同一 assembler + parser,落 treatment_completed
  - kind: union
    inputs: ['treatment_actual_rows', '_dispose_tx']
    output: treatment_actual_rows

  # ── C.9b dispose path 2:treat_plan 有真治疗但【全无牙位】→ 去 dispose 找回牙位 ──
  # 背景:医生在 treat_plan 写了治疗名(充填)却漏填 toothPosition,而牙位写进了 dispose
  #   (刘小源:treat_plan=[{treatName:"充填",toothPosition:""}],dispose 16;26 树脂充填)。
  #   path 1(blank_or_all_in)不命中(treat_plan 非空非复查)→ 牙位丢失 → 落无牙位充填,解不了任何牙。
  # gate:real_no_tooth —— 有真治疗(treatName ∉ review)且所有元素都无牙位。
  #   排除空/全复查(path 1 已处理,不重复抽 dispose 双计);排除任一元素已带牙位(treat_plan 牙位齐 → 不抽)。
  # ⭐ dispose 只留【带牙位】条目(tooth_position not_empty)→ 全口牙周/正畸(dispose 多无牙位)不被抽,
  #   避免与 treat_plan 全口治疗双计;只把结构治疗(充填@16;26)的牙位找回。host 录入习惯归 yaml。
  - kind: filter
    input: fact_emr_treatment_out
    output: _emr_dispose_gate2
    where:
      treat_plan: { real_no_tooth: *review_terms }
  - kind: split_json_array
    input: _emr_dispose_gate2
    output: _dispose_raw2
    array_field: dispose
    parent_keys:
      emr_id: id
      patient_id: patient_id
      organization_id: organization_id
      brand: brand
      rq: rq
      user_id: user_id
      doctor_name: doctor_name
      created_date: created_date
      updated_date: updated_date
    element_fields:
      treat_name: message
      tooth_position: toothPosition
    where:
      treat_name: { not_empty: true }
      tooth_position: { not_empty: true }   # ⭐ 只留带牙位的 dispose(限制双计)
  - kind: derive
    input: _dispose_raw2
    output: _dispose_tx2
    fields:
      treat_external_id:
        op: concat
        parts: ['${emr_id}', '|disp2|', '${tooth_position}', '|', '${treat_name}']
      category_raw:           # 归一(同 C.9/C.10):CJK 标点折叠 → strip 切段(建议/必要时/若…)生效
        op: normalize
        from: treat_name
  - kind: union
    inputs: ['treatment_actual_rows', '_dispose_tx2']
    output: treatment_actual_rows

  # ── C.10 ⭐ diag 里的"已治疗状态"(X术后/已充填)→ 补 actual 治疗【证据】 ──
  # 背景:医生 charting 常把"已补/已根管/已种植"等【已完成治疗的状态】记进 diag 字段
  #   (谢玉兰 24;25 = "充填术后")。它们 code=null,作为信号本就不召;
  #   但【必须 resolve 之前同牙的诊断】(如 2023 深龋),否则旧诊断永远无 resolver → 误召。
  #   ⚠️ 关键:不是"当噪音丢掉"(丢掉=旧深龋还在召),而是"认成已治疗"→ 去解决旧诊断。
  # 思路(复用 dispose 同款机器):message 当 treat_name → treatment_actual.yaml keyword_mapping
  #   (充填术后→restorative、根管术后→endodontic、种植术后→implant、拔牙术后→surgical)
  #   → 落 treatment_completed(occurredAt=本次到访)→ ≥ 之前诊断 → family resolver 自然解决。
  #   host charting 习惯归 yaml,召回算法零改动。
  - kind: route_by_pattern
    input: _diagnosis_raw
    field: message
    routes:
      - output: _post_treatment_raw
        when:
          # 已完成态(术后/已X/X后)+ 进行态(治疗中/术中/粘结中)都收:
          #   本场景=treatment_initiation_recall(未启动),"根管治疗中"=已启动 → 不该召"未启动"。
          #   ⚠️ 必须用【复合安全词】,避开裸"后/中"误伤(后牙/中切牙/不良修复体/下颌后缩)。
          #   "修复后"≠"不良修复体"(后 vs 体),特意不用裸"修复"。
          contains: ['术后', '已充填', '充填后', '治疗后', '治疗中', '术中',
                     '修复后', '备牙后', '拔牙后', '戴牙后', '取模后', '粘结中', '修整后']
      - output: _post_treatment_discard   # 其余诊断不在此处理(走上面 B → diagnosis_rows)
        when:
          default: true
  - kind: derive
    input: _post_treatment_raw
    output: _post_tx
    fields:
      treat_name:                # diag 的 message 当治疗名 → 交 keyword_mapping 含词分类
        op: normalize            # 对齐主路径 C.2.5:全角→半角 + 标点折叠(ＲＣＴ治疗后→RCT…)
        from: message
      treat_external_id:
        op: concat
        parts: ['${emr_id}', '|post|', '${tooth_position}', '|', '${message}']
      category_raw:              # ⭐ 真正喂分类器的字段(treatment_actual.yaml: category←category_raw)
        op: normalize            # 必须归一,否则全角/中标点变体 enum+keyword 全漏 → category 空 → 丢
        from: message
  - kind: union
    inputs: ['treatment_actual_rows', '_post_tx']
    output: treatment_actual_rows

  # ── E. EMR.file_url JSON 拆行 → 影像 metadata(image_record) ──
  # file_url 元素:{ check_name(影像类型), file_url(存储路径), created_gmt_at(拍摄时间) }
  # PAC 不持文件本体,只落 metadata:类型 / 时间 / 关联接诊 → "拍过 CBCT=种植意向" 类信号
  - kind: split_json_array
    input: fact_emr_treatment_out
    output: _image_raw
    array_field: file_url
    parent_keys:
      emr_id: id
      patient_id: patient_id
      organization_id: organization_id
      brand: brand
      registration_id: registration_id
      user_id: user_id           # 接诊医生(影像通常在接诊时拍)
      doctor_name: doctor_name
      created_date: created_date  # W4 末:DW upsert,传父级 created/updated_date
      updated_date: updated_date
    element_fields:
      check_name: check_name
      file_path: file_url
      captured_at: created_gmt_at
    where:
      file_path: { not_empty: true }

  # E.1 派生 image external_id(emr_id + 路径,跨数组元素保唯一)
  - kind: derive
    input: _image_raw
    output: image_rows
    fields:
      image_external_id:
        op: concat
        parts: ['${emr_id}', '|img|', '${file_path}']

  # ── D. settlement_out → refund_rows(退费)──
  #
  # W4 末:settlement → treatment_actual 路径已删除(详见 § C 顶部注释)
  # settlement 现在只服务两路:
  #   ① refund(本段)→ refund_rows → refund_record(LTV/风险算法)
  #   ② payment_mode(§ E)→ payment_rows → payment_record(LTV/渠道分析)
  #
  # 已治疗的"actual"信号源**只信任 EMR.treat_plan**(临床真实,带牙位),不再由 settlement 反推。
  #
  # SQL 已 WHERE is_refund=1 OR status=4;此处 project 准备 refund.yaml 消费。
  # refund.parser 侧 Math.abs(amount) 把 status=4 负 money 和 is_refund=1 正 money 都归一成"被退金额"正 cents。
  - kind: project
    input: fact_settlement_out_refund
    output: refund_rows
    keep: [id, organization_id, brand, patient_id, rq, billing_date, created_date,
           settlement_type, doctor_id, settlement_project_name, settlement_money,
           registration_id, is_refund, settlement_status]

  # ── E. settlement_mode_out → payment_rows ──
  # 17 列 pay_* → payment_channel 单列(代替 SQL multiIf)
  - kind: pick_first_nonzero
    input: fact_settlement_mode_out
    output: payment_rows
    output_field: payment_channel
    default: other
    candidates:
      - { column: pay_wechat,           value: wechat }
      - { column: pay_alipay,           value: alipay }
      - { column: pay_cash,             value: cash }
      - { column: pay_card,             value: card }
      - { column: pay_pos,              value: pos }
      - { column: pay_store,            value: store }
      - { column: pay_insurance,        value: insurance }
      - { column: pay_advance,          value: advance }
      - { column: pay_membershipcard,   value: membership_card }
      - { column: pay_medicalInsurance, value: medical_insurance }
      - { column: pay_apple,            value: apple_pay }
      - { column: pay_dsf,              value: third_party }
      - { column: pay_check,            value: check }
      - { column: pay_debt,             value: debt }
      - { column: pay_reserve,          value: mini_program }

  # ── F. 预约 → 拆"已到诊接诊事件"vs"全部预约"两张表 ──
  # F.1 已到诊(appo_status IN 4,5,6)
  - kind: route_by_pattern
    input: fact_appointment_out
    field: appo_status
    routes:
      - output: _appointment_arrived_raw
        when:
          in: ['4', '5', '6']
      - output: _non_arrived_rows
        when:
          default: true

  # F.2 进一步过滤:in_time 非空(SQL 朴素化后,这种行级 WHERE 用 filter operator 干)
  - kind: filter
    input: _appointment_arrived_raw
    output: appointment_encounter_rows
    where:
      in_time: { not_empty: true }

  # F.3 全部预约:拼精确预约时刻(appo_time 只到天,时分在 appo_datetime_str "13:00:00-14:00:00")
  #   appo_start_time = datetime_str 起始 8 位("13:00:00");scheduled_at_ts = 日期 + 起始时分
  #   datetime_str 为空时退化为 "YYYY-MM-DD "(尾空格,Date.parse 容忍)
  - kind: derive
    input: fact_appointment_out
    output: appointment_all_rows
    fields:
      appo_start_time:
        op: substring
        from: appo_datetime_str
        start: 0
        end: 8
      scheduled_at_ts:
        op: concat
        parts: ['${appo_time}', ' ', '${appo_start_time}']

  # ── G. fact_client_out → referral_rows(只取有 recommend_id 的行,即"被推荐来"的患者)──
  # fact_client_out 一行 = 一个患者;recommend_id 非空 → 该患者是被推荐来的 → 产 1 条 referral_record
  - kind: filter
    input: fact_client_out
    output: _referral_filtered
    where:
      recommend_id: { not_empty: true }   # null / 0 / '' 都过滤掉

  - kind: derive
    input: _referral_filtered
    output: referral_rows
    fields:
      # 推荐事件 id 合成:被推荐人 + 推荐人(同一被推荐人理论上只一条推荐关系)
      referral_external_id:
        op: concat
        parts: ['${patient_id}', '|ref|', '${recommend_id}']

  # ── H. fact_returnvisit_out:治疗项 大类(treatment_items)·子项(treatment_items_two)合一 ──
  #   host 把治疗项拆成两列(大类「外科」+ 子项「简单拔牙」),实测 5 试点两列总是成对(只填一列=0)。
  #   回访是纯展示,PAC 只存一个可读字段 treatmentItems —— "两列"的 host 怪癖在此合并,PAC schema 不变。
  - kind: derive
    input: fact_returnvisit_out
    output: fact_returnvisit_out
    fields:
      treatment_items_full:
        op: concat
        parts: ['${treatment_items}', ' · ', '${treatment_items_two}']

  # ── I. fact_consult_out:合成 consult_external_id(无 id 列)──
  #   形态改造(concat)归 transforms,SQL 保持朴素导出(CSV 双源等价)。
  #   (patient,appo,date) 全唯一(实测 368183 行 = uniq_composite)。
  - kind: derive
    input: fact_consult_out
    output: fact_consult_out
    fields:
      consult_external_id:
        op: concat
        parts: ['${patient_id}', '|', '${appo_id}', '|', '${appointment_date}']

# ─────────────────────────────────────────────────────────────
# PAC 写的 assembler 配置(每个 canonical resource 一份)
# ─────────────────────────────────────────────────────────────
assemblers:
  - { file: assemblers/patient.yaml }
  - { file: assemblers/patient_relation.yaml }   # 联系人/亲属边(referee → PatientRelation upsert)
  - { file: assemblers/patient_return_visit.yaml }  # 诊所回访(展示用,5 试点 → PatientReturnVisit upsert)
  - { file: assemblers/consult.yaml }               # 咨询主体(意向源,5 试点 → consultation_record)
  - { file: assemblers/encounter.yaml }
  - { file: assemblers/appointment.yaml }
  - { file: assemblers/diagnosis.yaml }
  - { file: assemblers/image_finding.yaml }   # 影像 AI 分析 → diagnosis_record(code_source=image_ai)
  - { file: assemblers/treatment_planned.yaml }
  - { file: assemblers/treatment_review.yaml }
  - { file: assemblers/treatment_actual.yaml }
  - { file: assemblers/recommendation.yaml }
  - { file: assemblers/emr.yaml }
  - { file: assemblers/image.yaml }
  # referral 暂不加载:jvs-dw 唯一可拼出 referral 的源是 fact_client_out(ADS 聚合,无 organization_id),
  #   而 PAC patient_transactions.clinic_id 是立柱必填(架构原则不放宽 — 不为单 host 凑活)。
  #   PAC referral_record 主体定义保留(schema / parser / canonical / yaml 都齐全),
  #   等其他 host 提供 DWD 明细级转介绍表(带 clinic_id)时启用。已记 DW 问题清单 #9。
  # - { file: assemblers/referral.yaml }
  - { file: assemblers/payment.yaml }
  - { file: assemblers/refund.yaml }
