#!/usr/bin/env bash
# FRIDAY SaaS 测试环境 → PAC cold-import 导出(host 侧「自洽业务表」参考实现)
#
# 纪律(2026-07 对齐,契约见 docs/integration/friday-push-payload):
#   - 业务 WHERE / status 切分 / K 码截取 / 数组拆行 → 全在 PAC transforms 层(不在导出侧)。
#   - 但**宿主自己的引用**在导出时 within-host join 后 inline 进相关行(PAC 无从做,数据在宿主手里):
#       私有字典码(diag linkCode→std_code、影像 class_code→类型名)、1:N 派生(默认联系号+归属)、
#       头-行属性(计划头 诊所/方案名 下放到计划行)。故不再导出纯字典表/联系方式/计划头/支付通道。
#   - **列名一律沿用宿主原表列名**(含 inline 进来的:contacts_tel/relationship/std_code/class_name/
#     organization_id/plan_name)—— PAC 不发明新名,宿主无需为 PAC 做字段改名。
#
# 用法(cohort 过滤,语义与 jvs-dw cold-import --clinics/--since 完全一致):
#   ./export.sh                                    # 全量(默认行为不变)
#   ./export.sh --clinics=<orgId,orgId>            # 只导"在这些诊所看过(有正式病历)"的患者
#   ./export.sh --since=YYYY-MM-DD                 # 只导"该日期后有来诊"的患者
#   ./export.sh --months=N                         # 最近 N 个月有来诊(换算成 --since;--since 显式优先)
#   ⚠️ 过滤只收窄「患者名单」(cohort 源 = med_emr_info 正式病历,对齐 jvs-dw clinic_scope
#     挂 EMR 事实表):入选患者的**全部主体、跨所有诊所、全部历史**照常导出,不按行截断。
#     --clinics 与 --since 可叠加(按患者求交,非行级 AND)。导入幂等追加,分批补摄安全。
#
# 凭据走 env(不入 git):
#   export FRIDAY_MYSQL_HOST=… FRIDAY_MYSQL_PORT=3309 FRIDAY_MYSQL_USER=… FRIDAY_MYSQL_PASSWORD=…
#   export FRIDAY_MONGO_URI='mongodb://user:pass@host:27017/emr_market'   # 密码含 & 等须 URL 编码
#
# 依赖:docker(mysql:8 / mongo:7 客户端镜像),产物落本目录(.gitignore 已挡,不入 git)。
set -euo pipefail
cd "$(dirname "$0")"

# ── cohort 过滤参数(在凭据校验前解析:--help 不需要凭据)──
CLINICS="" SINCE="" MONTHS=""
for a in "$@"; do
  case "$a" in
    --clinics=*) CLINICS="${a#--clinics=}" ;;
    --since=*)   SINCE="${a#--since=}" ;;
    --months=*)  MONTHS="${a#--months=}" ;;
    --help|-h)   sed -n '/^# 用法/,/^#$/p' "$0"; exit 0 ;;
    *) echo "未知参数: $a(支持 --clinics= --since= --months=)" >&2; exit 1 ;;
  esac
done
# --since 显式优先;否则 --months=N 换算(BSD date 优先,GNU date 兜底)
if [ -z "$SINCE" ] && [ -n "$MONTHS" ]; then
  SINCE=$(date -v-"${MONTHS}"m +%F 2>/dev/null || date -d "-${MONTHS} months" +%F)
fi

: "${FRIDAY_MYSQL_HOST:?need FRIDAY_MYSQL_HOST}"
: "${FRIDAY_MYSQL_PORT:=3309}"
: "${FRIDAY_MYSQL_USER:?need FRIDAY_MYSQL_USER}"
: "${FRIDAY_MYSQL_PASSWORD:?need FRIDAY_MYSQL_PASSWORD}"
: "${FRIDAY_MONGO_URI:?need FRIDAY_MONGO_URI}"

COHORT_IN=""   # 非空 = cohort 模式:MySQL 谓词 "IN ('id',...)"
if [ -n "$CLINICS$SINCE" ]; then
  echo "── cohort 计算(med_emr_info:clinics=${CLINICS:-*} since=${SINCE:-*})──"
  # 谓词按「患者」求交(与 jvs-dw 同):在指定诊所看过 ∩ 该日期后有来诊;各谓词独立,非同行 AND。
  docker run --rm -e "CL=$CLINICS" -e "SN=$SINCE" mongo:7 mongosh "$FRIDAY_MONGO_URI" --quiet --eval '
    const cl = (process.env.CL || "").split(",").map((s) => s.trim()).filter(Boolean);
    const sn = process.env.SN || "";
    const base = { status: { $in: [3, 4] } };
    // 存储类型 string/number 混杂可能 → 同值双形态都进 $in
    const both = (s) => { const n = Number(s); return Number.isFinite(n) && String(n) === s ? [s, n] : [s]; };
    const S = (v) => String(v);
    let ids = null;
    if (cl.length) ids = db.med_emr_info.distinct("patientId", Object.assign({ organizationId: { $in: cl.flatMap(both) } }, base)).map(S);
    if (sn) {
      // clinicTime 是"北京墙钟伪装成 UTC"的 Date(见下方 EMR 导出注释)→ cutoff 按 Z 构造即对齐存储
      const set = new Set(db.med_emr_info.distinct("patientId", Object.assign({ clinicTime: { $gte: new Date(sn + "T00:00:00Z") } }, base)).map(S));
      ids = ids === null ? Array.from(set) : ids.filter((x) => set.has(x));
    }
    print(JSON.stringify(ids ?? []));
  ' > cohort_ids.json.tmp
  node -e "
    let a; try { a = JSON.parse(require('fs').readFileSync('cohort_ids.json.tmp','utf-8')); } catch (e) {
      console.error('  ✗ cohort 计算失败(Mongo 不可达?)'); process.exit(1); }
    if (!Array.isArray(a) || a.length === 0) {
      console.error('  ✗ 过滤后 cohort 为空(诊所 id 拼错?日期太晚?)— 拒绝继续(否则会退化成全量导出)'); process.exit(1); }
    console.log('  cohort:', a.length, '患者');
  " || { rm -f cohort_ids.json.tmp; exit 1; }
  mv cohort_ids.json.tmp cohort_ids.json
  COHORT_IN=$(node -e '
    const ids = JSON.parse(require("fs").readFileSync("cohort_ids.json", "utf-8"));
    process.stdout.write("(" + ids.map((s) => "\x27" + String(s).replace(/\x27/g, "") + "\x27").join(",") + ")");
  ')
fi

# cohort 谓词拼接:pf = 查询无 WHERE 时用;pfa = 已有 WHERE 时接 AND。无 cohort → 空串(全量不变)。
pf()  { if [ -n "$COHORT_IN" ]; then printf 'WHERE %s IN %s' "$1" "$COHORT_IN"; fi; }
pfa() { if [ -n "$COHORT_IN" ]; then printf 'AND %s IN %s' "$1" "$COHORT_IN"; fi; }
# settlement_modes 无患者列 → 经结算头单挂靠(IN 子查询是 jvs-dw SQL 纪律明确放行的 cohort 限定形态)
pfs() { if [ -n "$COHORT_IN" ]; then printf 'WHERE settlement_id IN (SELECT uuid FROM `arrail-settlement-server`.patient_settlement WHERE patient_id IN %s)' "$COHORT_IN"; fi; }

# ── MySQL → CSV(TSV 中转,node 处理 mysql --batch 的 \t \n \\ 转义再按 RFC4180 引号)──
mysql_csv() {
  local sql="$1" out="$2"
  # SQL 走 stdin 而非 -e:cohort IN 列表可达数十万字符,作为 docker 参数会撞 ARG_MAX
  printf '%s\n' "$sql" | docker run -i --rm mysql:8 mysql \
    -h"$FRIDAY_MYSQL_HOST" -P"$FRIDAY_MYSQL_PORT" -u"$FRIDAY_MYSQL_USER" -p"$FRIDAY_MYSQL_PASSWORD" \
    --default-character-set=utf8mb4 --batch --raw=FALSE --connect-timeout=20 \
    | node -e '
      const lines = require("fs").readFileSync(0, "utf-8").split("\n");
      const unesc = (s) => { let r = ""; for (let i = 0; i < s.length; i++) { const c = s[i];
        if (c === "\\" && i + 1 < s.length) { const n = s[++i];
          r += n === "n" ? "\n" : n === "t" ? "\t" : n === "0" ? "\0" : n === "N" ? "" : n; }
        else r += c; } return r; };
      const q = (s) => /[",\n\r]/.test(s) ? `"${s.replace(/"/g, `""`)}"` : s;
      const out = [];
      for (const line of lines) { if (line === "") continue;
        out.push(line.split("\t").map((f) => q(f === "NULL" ? "" : unesc(f))).join(",")); }
      process.stdout.write(out.join("\n") + "\n");
    ' > "$out.tmp"
  # ⚠️ 先写 .tmp,校验有数据才替换正式文件 —— 测试库抖动时裸重定向会把好文件清空(2026-07-20 踩)
  local n=$(($(wc -l < "$out.tmp") - 1))
  if [ "$n" -lt 1 ]; then
    # cohort 模式下 0 行是合法结果(该批患者恰无此类数据;mysql --batch 空结果连表头都不出)
    # → 从 SELECT 列表合成表头,产出 0 行 CSV(cold-import 仍需列名行)。全量模式维持拒绝覆盖。
    if [ -n "$COHORT_IN" ]; then
      printf '%s\n' "$sql" | sed -E 's/^SELECT ([^ ]+) FROM .*$/\1/' > "$out.tmp"
      mv "$out.tmp" "$out"
      echo "  $out: 0 rows(cohort 内无此类数据,合成表头)"
      return 0
    fi
    rm -f "$out.tmp"
    echo "  ✗ $out: 导出为空(库不可达?),保留原文件不覆盖" >&2
    return 1
  fi
  mv "$out.tmp" "$out"
  echo "  $out: $n rows"
}

echo "── MySQL 导出 ──"
# ⭐ 自洽形态:宿主 within-host join 把私有字典码 / 1:N 默认号 / 计划头属性 inline 进相关表
#   (契约见 docs/integration/friday-push-payload)。纯字典表 / 联系方式 1:N / 计划头 / 支付通道
#   不再单独导出成文件——PAC 只摄入自洽业务行(所有摄入路径同形态)。
# 患者主档 + inline 默认联系号(挑:非空号 → is_default↓ → contacts_type↑ → id↑;phone 与归属同源同一条)
mysql_csv "SELECT cbi.id,cbi.name,cbi.sex,cbi.birthday,cbi.file_number,cbi.tenant_id,cbi.organization_id,cbi.created_gmt_at,cbi.updated_gmt_at,
  (SELECT c.contacts_tel  FROM customer.customer_contacts c WHERE c.customer_id=cbi.id AND c.contacts_tel<>'' ORDER BY c.is_default DESC,c.contacts_type ASC,c.id ASC LIMIT 1) AS contacts_tel,
  (SELECT c.relationship  FROM customer.customer_contacts c WHERE c.customer_id=cbi.id AND c.contacts_tel<>'' ORDER BY c.is_default DESC,c.contacts_type ASC,c.id ASC LIMIT 1) AS relationship
  FROM customer.customer_basic_info cbi $(pf cbi.id)" customer_basic_info.csv
# 预约(全状态;transforms 丢草稿 10)
mysql_csv "SELECT id,patient_appointment_id,organization_id,tenant_id,appointment_date,appointment_start,appointment_status,doctor_user_id,appointment_time_length,in_time,created_gmt_at,updated_gmt_at FROM \`arrail-appointment-server\`.appointment_base $(pf patient_appointment_id)" appointment_base.csv
# 影像档案 + inline class_name(class_code→std_check_class,宿主解析;PAC 再翻 modality)
mysql_csv "SELECT mc.id,mc.tenant_id,mc.organization_id,mc.organization_name,mc.patient_id,mc.emr_id,mc.class_code,sc.class_name,mc.file_name,mc.file_type,mc.file_url,mc.files_size,mc.shooting_time,mc.created_gmt_at,mc.updated_gmt_at FROM emr.med_check mc LEFT JOIN emr.std_check_class sc ON sc.class_code=mc.class_code $(pf mc.patient_id)" med_check.csv
# 治疗计划行 + inline 头属性(organization_id/plan_name 取自计划头;头不单独导出)
mysql_csv "SELECT it.id,it.tenant_id,it.customer_id,it.treat_plan_id,it.tooth_position,it.mode_name,it.charge_min,it.charge_max,it.created_gmt_at,it.updated_gmt_at,h.organization_id,h.plan_name FROM customer.customer_treat_plan_item it LEFT JOIN customer.customer_treat_plan h ON h.id=it.treat_plan_id $(pf it.customer_id)" customer_treat_plan_item.csv
# 结算头单原表(全 status,导出侧不做业务 WHERE;消费/退费切分全在 PAC)
mysql_csv "SELECT uuid,tenant_id,organization_id,patient_id,doctor_id,status,receivable_this,net_receipts_this,billing_date,registration_id,ref_settlement_id,settlement_serial_num,reason,created_gmt_at,updated_gmt_at FROM \`arrail-settlement-server\`.patient_settlement $(pf patient_id)" patient_settlement.csv
# 结算明细原表(全量含 is_refund,PAC 侧切退费明细)。JOIN 结算头:① 挂靠限定 cohort
#   ② 反填真 patient_id(spec.patient_id 部分品牌是诊所本地 id,不可信 → 取结算头 ps.patient_id inline)
mysql_csv "SELECT sp.id,sp.tenant_id,sp.organization_id,ps.patient_id,sp.settlement_id,sp.cure_name,sp.service_project_name,sp.receivable_this,sp.net_receipts_this,sp.is_refund,sp.created_gmt_at,sp.updated_gmt_at FROM \`arrail-settlement-server\`.patient_settlement_spec sp JOIN \`arrail-settlement-server\`.patient_settlement ps ON ps.uuid=sp.settlement_id $(pf ps.patient_id)" patient_settlement_spec.csv
# 转介绍圈(患者-患者关系边,双向成对存;referee_relationship 为宿主字典 id,PAC 侧统计推断解码)
mysql_csv "SELECT id,tenant_id,organization_id,customer_id,referee_patient_id,referee_relationship,type,created_gmt_at,updated_gmt_at FROM customer.customer_referee_circle $(pf customer_id)" customer_referee_circle.csv
# 咨询(意向 potential_treatment/未成交原因;测试库仅 45 行,结构真实)
mysql_csv "SELECT id,tenant_id,organization_id,customer_id,task_director_id,task_director,treatment_plan,preferential_way,unsettled_reason,unsettled_reason_description,other_conditions,potential_treatment,data_sources,created_gmt_at,updated_gmt_at FROM customer.customer_consult $(pf customer_id)" customer_consult.csv

echo "── Mongo 导出(med_emr_info,status∈{3,4} 正式病历)──"
# 不用 mongoexport:Extended JSON 会把 Date/Long 包成 {$date}/{$numberLong} 对象,破坏平铺契约。
# mongosh 手动平铺:Date→naive 北京墙钟字符串,Long→字符串,数组(diag/treat/dispose/examine)原样保留
# (PAC split_json_array 原生支持已 parse 数组)。产出 JSON 数组文件(cold-import json 格式)。
#
# ⚠️ Date 时区实证(2026-07-20,clinicTime/createdGmtAt 小时分布双证):FRIDAY 写库时把
#   北京墙钟"伪装成 UTC"存进 Mongo Date(直读 UTC 字段=营业时间双峰 上午7-10/下午13-17;
#   按真 UTC 解释则一半病历落深夜——荒谬)。故导出**不能** toISOString 带 Z(会被 PAC 当真
#   UTC 再 +8,EMR 链时间全偏 8 小时,已踩):把 UTC 字段值直读为墙钟,输出 naive
#   "YYYY-MM-DD HH:mm:ss",交给 manifest timezone=Asia/Shanghai 解释 → 瞬间正确。
# 诊断字典 map(diag_code→std_code)—— diag[] inline std_code 的 within-host join;linkCode 在 Mongo、
#   字典在 MySQL,故预取成 map 挂进 mongo 容器,eval 里逐元素解析(不再 PAC 侧 lookup std_diag)。
printf 'SELECT diag_code,std_code FROM emr.std_diag\n' | docker run -i --rm mysql:8 mysql \
  -h"$FRIDAY_MYSQL_HOST" -P"$FRIDAY_MYSQL_PORT" -u"$FRIDAY_MYSQL_USER" -p"$FRIDAY_MYSQL_PASSWORD" \
  --default-character-set=utf8mb4 --batch --raw=FALSE --connect-timeout=20 \
  | node -e '
    const lines = require("fs").readFileSync(0, "utf-8").split("\n"); const m = {};
    for (let i = 1; i < lines.length; i++) { const l = lines[i]; if (!l) continue; const t = l.split("\t");
      if (t[0] && t[0] !== "NULL") m[t[0]] = (t[1] && t[1] !== "NULL") ? t[1] : ""; }
    process.stdout.write(JSON.stringify(m));
  ' > std_diag_map.json
echo "  std_diag_map.json: $(node -e 'process.stdout.write(String(Object.keys(JSON.parse(require("fs").readFileSync("std_diag_map.json","utf-8"))).length))') 码"

# cohort 模式:挂载 id 文件供 $in 过滤(bash 3.2 + set -u 下空数组要用 ${arr[@]+...} 展开)
MONGO_ARGS=(-v "$PWD/std_diag_map.json:/std_diag_map.json:ro")
if [ -n "$COHORT_IN" ]; then MONGO_ARGS+=(-v "$PWD/cohort_ids.json:/cohort_ids.json:ro" -e "COHORT=1"); fi
docker run --rm ${MONGO_ARGS[@]+"${MONGO_ARGS[@]}"} mongo:7 mongosh "$FRIDAY_MONGO_URI" --quiet --eval '
  const flat = (v) => v instanceof Date ? v.toISOString().slice(0, 19).replace("T", " ")
    : (v && v.constructor && v.constructor.name === "Long") ? String(v)
    : v === undefined ? null : v;
  // id 类字段字符串化:Mongo 存的是 number/Long 混型,canonical 契约要 string
  // (CSV 源天然全字符串;JSON 导出须显式对齐,否则 patientExternalId 等校验失败)
  const sid = (v) => (v === undefined || v === null || v === "") ? null : String(flat(v));
  // 诊断字典 map(within-host join):diag[].linkCode → std_code inline 进元素(沿用宿主列名 std_code)
  let DIAG = {}; try { DIAG = JSON.parse(require("fs").readFileSync("/std_diag_map.json", "utf-8")); } catch (e) {}
  const q = { status: { $in: [3, 4] } };
  try {
    // cohort 激活但文件读不到 → 硬失败(宁可让 .tmp 校验拒绝,不能静默退化成全量导出)
    const ids = JSON.parse(require("fs").readFileSync("/cohort_ids.json", "utf-8"));
    const both = (s) => { const n = Number(s); return Number.isFinite(n) && String(n) === s ? [s, n] : [s]; };
    q.patientId = { $in: ids.flatMap(both) };
  } catch (e) {}
  let first = true;
  print("[");
  db.med_emr_info.find(q).forEach((d) => {
    const o = {
      emr_sub_id: sid(d.emrSubId), emr_id: sid(d.emrId),
      appointment_id: sid(d.appointmentId),
      tenant_id: sid(d.tenantId), organization_id: sid(d.organizationId),
      organization_name: flat(d.organizationName),   // 诊所名(clinic_directory 派生 host.clinicNames)
      patient_id: sid(d.patientId), user_id: sid(d.userId), user_name: flat(d.userName),
      clinic_time: flat(d.clinicTime), visit_indicator: flat(d.visitIndicator), status: flat(d.status),
      // diag[] inline std_code(宿主 within-host join);treat/dispose/examine 原生数组原样
      diag: (d.diag ?? []).map((el) => Object.assign({}, el,
        { std_code: (el && el.linkCode != null && el.linkCode !== "") ? (DIAG[String(el.linkCode)] ?? "") : "" })),
      treat: d.treat ?? [], dispose: d.dispose ?? [], examine: d.examine ?? [],
      // emr_record 自由文本位要字符串(fact 层 zod nullableString);数组原样给 split_json_array 拆行
      dispose_text: (d.dispose ?? []).length ? JSON.stringify(d.dispose) : null,
      examine_text: (d.examine ?? []).length ? JSON.stringify(d.examine) : null,
      illness_desc: flat(d.illnessDesc), pre_illness: flat(d.preIllness), past_hist: flat(d.pastHist),
      gen_cond: flat(d.genCond), auxi_exam: flat(d.auxiExam), doc_order: flat(d.docOrder),
      created_gmt_at: flat(d.createdGmtAt), updated_gmt_at: flat(d.updatedGmtAt),
    };
    print((first ? "" : ",") + JSON.stringify(o)); first = false;
  });
  print("]");
' > med_emr_info.json.tmp
# 同款保护:解析成功且非空才替换(mongosh 连接失败会产出空/半截文件)
node -e "
  const fs=require('fs');
  let a; try { a = JSON.parse(fs.readFileSync('med_emr_info.json.tmp','utf-8')); } catch (e) {
    fs.unlinkSync('med_emr_info.json.tmp');
    console.error('  ✗ med_emr_info.json: 导出无效(库不可达?),保留原文件不覆盖'); process.exit(1);
  }
  if (!Array.isArray(a) || a.length === 0) {
    fs.unlinkSync('med_emr_info.json.tmp');
    console.error('  ✗ med_emr_info.json: 导出为空,保留原文件不覆盖'); process.exit(1);
  }
  fs.renameSync('med_emr_info.json.tmp','med_emr_info.json');
  console.log('  med_emr_info.json:', a.length, 'docs');
"

echo "── 导出完成 ──"
ls -la *.csv *.json | awk '{print "  " $NF, $5}'
