# 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
        )

# ─────────────────────────────────────────────────────────────
# 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
      std_code_k:
        op: substring
        from: std_code
        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=planned,本次做的复查,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 形态(中标点已折叠);不要在这里写中括号/中逗号
      - output: _treatment_review_raw
        when:
          equals:
            # 复查 / 检查 / 复诊
            - 常规复查
            - 复查
            - 定期复查
            - 检查
            - 初诊检查
            - 正畸复查
            - 正畸检查
            - 正畸复诊
            - 种植复查
            - 牙周复查
            - 保持器复诊
            # 暂观 / 观察 / 拆线
            - 拆线
            - 暂观
            - 观察
            - 观察随访
            - 随诊观察
            - 随访观察
            - 无治疗
            - 重复
            - 取资料
            - 暂观,必要时拔除
            - 观察,必要时拔除
            - 随访观察,必要时拔除
            # 流程 / 咨询 / 行政
            - 转诊
            - 请全科医生会诊
            - 正畸咨询
            - 正畸会诊
            - 方案沟通
            - 沟通治疗方案
            - 听方案
            - 缴费
            - 未拍片
            - 拒绝拍片
            - 治疗中
            - 已交付纸质病历
      # 真治疗动作 → treatment_actual_rows ⭐ kind=actual(临床真实)
      - output: _treatment_actual_raw_emr
        when:
          default: true

  # ── C.4 plan 路由分流(MVP:只取真治疗,推荐/复查暂 drop)──
  - kind: route_by_pattern
    input: _plan_raw
    field: treat_name
    routes:
      # MVP:plan 的"建议"暂时 drop(transforms 无 union,合并 treat_plan 留 TODO)
      - output: _drop_plan_recommendation
        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.5 推荐(来自 treat_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: ''

  # ── 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}']

# ─────────────────────────────────────────────────────────────
# PAC 写的 assembler 配置(每个 canonical resource 一份)
# ─────────────────────────────────────────────────────────────
assemblers:
  - { file: assemblers/patient.yaml }
  - { file: assemblers/encounter.yaml }
  - { file: assemblers/appointment.yaml }
  - { file: assemblers/diagnosis.yaml }
  - { 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 }
