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

# ── 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:
    url: http://cc-2zen8w29e49076o83.public.clickhouse.ads.aliyuncs.com:8123
    database: dw_group
    username: jvs_pac
    password_env: DW_CLICKHOUSE_PASSWORD
  # 1000 cohort 患者 settlement 总行 ~ 20 万(p50=12/患者,max=11.6 万/异常患者)
  # 100k 会截一半数据,关键事实(吴建康 61 行种植结算)被丢失 → scenario 错召回
  # W3 末 demo 调到 500k,长期(生产)走增量 pull(cursor)替代一次性拉
  default_limit: 500000

  # SQL 最朴素化 — host(DW)给的数据范围就是 PAC 该消化的范围。
  # PAC 这边不预设诊所 / brand / 时间过滤,数据来什么就是什么。
  # 仅保留 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
      ORDER BY last_visit_time DESC
      LIMIT 1000

    # ── 预约(全部状态;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
        ORDER BY last_visit_time DESC LIMIT 1000
      )

    # ── EMR 全字段(自由文本 + diag/treat_plan JSON,后续 transforms 拆) ──
    fact_emr_treatment_out: |
      SELECT
        id, patient_id, organization_id, brand, rq, registration_id,
        illness_desc, pre_illness, past_hist, examine, dispose, doc_order,
        diag, treat_plan, plan, file_url
      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
        ORDER BY last_visit_time DESC LIMIT 1000
      )

    # ── 结算明细(已做治疗反推;退费/未完成结算行 PAC 不消费) ──
    fact_settlement_out: |
      SELECT id, organization_id, brand, patient_id, rq, settlement_type,
             settlement_project_name, settlement_money, 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
          ORDER BY last_visit_time DESC LIMIT 1000
        )

    # ── 支付通道明细(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
          ORDER BY last_visit_time DESC LIMIT 1000
        )

# ─────────────────────────────────────────────────────────────
# 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
      - illness_desc
      - pre_illness
      - past_hist
      - examine
      - dispose
      - doc_order

  # ── 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
    element_fields:
      std_code: stdCode
      message: message
      tooth_position: toothPosition
    where:
      message: { not_empty: true }    # 只要求有诊断名(message);stdCode 可空

  # B.1 归一 message(去尾部标点/空白)+ stdCode 截短取 K 码
  - kind: derive
    input: _diagnosis_raw
    output: _diagnosis_norm
    fields:
      message_norm:
        op: trim
        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 JSON 拆行 → 治疗 / 推荐 / 流程性 三路分流 ──
  - 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
    element_fields:
      treat_name: treatName
      tooth_position: toothPosition
      plan_code: planCode
    where:
      treat_name: { not_empty: true }

  # C.1 按 treat_name 关键词分流到 3 张表
  - kind: route_by_pattern
    input: _treat_plan_raw
    field: treat_name
    routes:
      # 推荐建议(医生说"建议XX")→ recommendation_record(W5+ 还会接 Layer C LLM 抽取增强)
      - output: _recommendation_raw
        when:
          starts_with: ['建议', '推荐']
      # 流程性事实(复查 / 拆线 / 观察 / 无治疗 / 重复 等)— W3 末加 review category
      - output: _treatment_review_raw
        when:
          equals:
            - 常规复查
            - 复查
            - 定期复查
            - 检查
            - 初诊检查
            - 拆线
            - 暂观
            - 观察
            - 无治疗
            - 重复
            - 取资料
            - 暂观,必要时拔除
      # 真治疗动作 → treatment_record(planned)
      - output: _treatment_planned_raw
        when:
          default: true

  # C.2 推荐:派生 external_id + 暂时把 treat_name 当作"推荐码"原文(后续 yaml enum_mapping 翻译)
  - kind: derive
    input: _recommendation_raw
    output: recommendation_rows
    fields:
      rec_external_id:
        op: concat
        parts: ['${emr_id}', '|rec|', '${treat_name}', '|', '${tooth_position}']

  # C.3 复查类:treat_name 当 subtype,category 全部 'review'(yaml 里 enum_mapping 处理)
  - 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.4 真治疗:derive external_id,category_raw 留 treat_name 原文(yaml enum_mapping 翻译)
  - kind: derive
    input: _treatment_planned_raw
    output: treatment_planned_rows
    fields:
      treat_external_id:
        op: concat
        parts: ['${emr_id}', '|tx|', '${treat_name}', '|', '${tooth_position}']
      category_raw:
        op: default
        from: treat_name
        value: ''

  # ── D. settlement_out → treatment_actual_rows ──
  # SQL 已 WHERE is_refund=0 + status=1;此处不再拆,直接 project rename 准备 enum_mapping
  - kind: project
    input: fact_settlement_out
    output: _treatment_actual_raw
    keep: [id, organization_id, brand, patient_id, rq, settlement_type,
           settlement_project_name, settlement_money, registration_id]
    rename:
      settlement_id: id    # 跟下游 yaml field_mapping 对齐

  # D.1 settlement_type 当 category_raw,yaml enum_mapping 翻译到 PAC category(45 个枚举)
  - kind: derive
    input: _treatment_actual_raw
    output: treatment_actual_rows
    fields:
      category_raw:
        op: default
        from: settlement_type
        value: ''

  # ── 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 全部预约:project 直通(其实可以直接消费 fact_appointment_out;这里复制一份给 yaml 清晰)
  - kind: project
    input: fact_appointment_out
    output: appointment_all_rows
    # keep 不写 = 全保留

# ─────────────────────────────────────────────────────────────
# 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/payment.yaml }
