feat(friday): export.sh 支持 --clinics/--since/--months cohort 过滤 — 语义对齐 jvs-dw

文件源宿主的诊所/时间过滤放导出侧(架构约定:host 端 dump 时过滤,PAC 不动):
- cohort 源 = med_emr_info 正式病历(对齐 jvs-dw clinic_scope 挂 EMR 事实表);
  只收窄患者名单,入选患者全部主体/跨诊所/全历史照常导出,谓词按患者求交非行级 AND
- settlement_modes 无患者列经结算头单 IN 子查询挂靠(纪律放行形态)
- SQL 走 stdin(cohort IN 列表会撞 ARG_MAX);空 cohort 拒跑防退化全量;
  cohort 内 0 行表合成表头(mysql --batch 空结果连表头都不出)
- cold-import 文件源传 --clinics/--since 时显式警告(此前静默忽略)
- refresh-clinic-names 改合并语义:子集导出不再刷掉子集外诊所名

实测(刘医生演示诊所 36 患者):EMR 53 份含 4 家其他诊所历史,零越界;
dry-run 15 资源 746 txns 0 failed;空 cohort exit=1;0 行表合成表头可导入。

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

apps/pac-service/data/friday/export.sh

@@ -5,6 +5,15 @@
 #   (2026-07 对齐:一列都不改名 —— 宿主原生字段直出;与 PAC 租户概念的区分由 PAC 适配层消化。)
 #   形态改造(拆 JSON 数组/字典 join/K 码截取)全部在 PAC transforms 层做。
 #
+# 用法(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 编码
@@ -13,18 +22,75 @@
 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"
-  docker run --rm mysql:8 mysql \
+  # 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 -e "$sql" \
+    --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];
@@ -40,6 +106,14 @@ mysql_csv() {
   # ⚠️ 先写 .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
@@ -50,34 +124,34 @@ mysql_csv() {
 
 echo "── MySQL 导出 ──"
 # 患者主档(全量;既有 450 行样本 → 扩全量,EMR 覆盖 ~1.65 万患者的前置)
-mysql_csv "SELECT id,name,sex,birthday,file_number,tenant_id,organization_id,created_gmt_at,updated_gmt_at FROM customer.customer_basic_info" customer_basic_info.csv
+mysql_csv "SELECT id,name,sex,birthday,file_number,tenant_id,organization_id,created_gmt_at,updated_gmt_at FROM customer.customer_basic_info $(pf id)" customer_basic_info.csv
 # 联系方式 1:N(lookup 挑默认号 → patient.phone)
-mysql_csv "SELECT id,customer_id,contacts_tel,is_default,contacts_type,tel_type,relationship FROM customer.customer_contacts" customer_contacts.csv
+mysql_csv "SELECT id,customer_id,contacts_tel,is_default,contacts_type,tel_type,relationship FROM customer.customer_contacts $(pf customer_id)" customer_contacts.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" appointment_base.csv
+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
 # 影像档案(挂 emr_id 病历号;modality 经 std_check_class 字典翻)
-mysql_csv "SELECT id,tenant_id,organization_id,organization_name,patient_id,emr_id,class_code,file_name,file_type,file_url,files_size,shooting_time,created_gmt_at,updated_gmt_at FROM emr.med_check" med_check.csv
+mysql_csv "SELECT id,tenant_id,organization_id,organization_name,patient_id,emr_id,class_code,file_name,file_type,file_url,files_size,shooting_time,created_gmt_at,updated_gmt_at FROM emr.med_check $(pf patient_id)" med_check.csv
 # 影像类型字典(class_code UUID 全局唯一,lookup 不需 tenant 限定)
 mysql_csv "SELECT class_code,class_name,tenant_id FROM emr.std_check_class" std_check_class.csv
 # 诊断字典(linkCode → std_code;K 码用,非 K 落自由文本)
 mysql_csv "SELECT diag_code,diag_name,std_code FROM emr.std_diag" std_diag.csv
 # 治疗计划(头:方案名/期望;行:牙位/术式/价格区间)→ treatment_planned
-mysql_csv "SELECT id,tenant_id,organization_id,customer_id,plan_group_id,plan_name,desired_effect,emergency_urgency,expect_cost,created_gmt_at,updated_gmt_at FROM customer.customer_treat_plan" customer_treat_plan.csv
-mysql_csv "SELECT id,tenant_id,customer_id,treat_plan_id,tooth_position,mode_name,charge_min,charge_max,created_gmt_at,updated_gmt_at FROM customer.customer_treat_plan_item" customer_treat_plan_item.csv
+mysql_csv "SELECT id,tenant_id,organization_id,customer_id,plan_group_id,plan_name,desired_effect,emergency_urgency,expect_cost,created_gmt_at,updated_gmt_at FROM customer.customer_treat_plan $(pf customer_id)" customer_treat_plan.csv
+mysql_csv "SELECT id,tenant_id,customer_id,treat_plan_id,tooth_position,mode_name,charge_min,charge_max,created_gmt_at,updated_gmt_at FROM customer.customer_treat_plan_item $(pf customer_id)" customer_treat_plan_item.csv
 # 结算(消费/退费)。status 语义(2026-07-18 实测):1=已结算 3=含退费行的单 4=整单反向冲减(负额+ref 挂原单)。
 # ⚠️ 2ac96f6d 品牌不用 status=4:整单冲减直接记 status=3 负金额(实测 200 行)→ 按金额正负切分:
 #   消费 = status∈{1,3} 且金额≥0;整单退费 = status=4 或 status=3 负金额(退费第三表达)。
 # 其余 status(0 草稿/2/5/6/7/8)语义未明,先不收 → WHERE 收窄(host 等价 dump 允许的过滤)
-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 WHERE status IN (1,3) AND receivable_this >= 0" patient_settlement.csv
-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 WHERE status=4 OR (status=3 AND receivable_this < 0)" patient_settlement_refund.csv
+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 WHERE status IN (1,3) AND receivable_this >= 0 $(pfa patient_id)" patient_settlement.csv
+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 WHERE (status=4 OR (status=3 AND receivable_this < 0)) $(pfa patient_id)" patient_settlement_refund.csv
 # 退费行级明细(部分退费轨道:is_refund=1 全部归属 status=3 头单)
-mysql_csv "SELECT id,tenant_id,organization_id,patient_id,settlement_id,cure_name,service_project_name,receivable_this,net_receipts_this,is_refund,created_gmt_at,updated_gmt_at FROM \`arrail-settlement-server\`.patient_settlement_spec WHERE is_refund=1" patient_settlement_spec_refund.csv
+mysql_csv "SELECT id,tenant_id,organization_id,patient_id,settlement_id,cure_name,service_project_name,receivable_this,net_receipts_this,is_refund,created_gmt_at,updated_gmt_at FROM \`arrail-settlement-server\`.patient_settlement_spec WHERE is_refund=1 $(pfa patient_id)" patient_settlement_spec_refund.csv
 # 支付通道长表(每单×通道一行;lookup 挑金额最大者为主导通道)
-mysql_csv "SELECT id,tenant_id,settlement_id,modes_name,money,created_gmt_at,updated_gmt_at FROM \`arrail-settlement-server\`.settlement_modes" settlement_modes.csv
+mysql_csv "SELECT id,tenant_id,settlement_id,modes_name,money,created_gmt_at,updated_gmt_at FROM \`arrail-settlement-server\`.settlement_modes $(pfs)" settlement_modes.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" customer_referee_circle.csv
+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" customer_consult.csv
+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} 对象,破坏平铺契约。
@@ -89,16 +163,26 @@ echo "── Mongo 导出(med_emr_info,status∈{3,4} 正式病历)──"
 #   按真 UTC 解释则一半病历落深夜——荒谬)。故导出**不能** toISOString 带 Z(会被 PAC 当真
 #   UTC 再 +8,EMR 链时间全偏 8 小时,已踩):把 UTC 字段值直读为墙钟,输出 naive
 #   "YYYY-MM-DD HH:mm:ss",交给 manifest timezone=Asia/Shanghai 解释 → 瞬间正确。
-docker run --rm mongo:7 mongosh "$FRIDAY_MONGO_URI" --quiet --eval '
+# cohort 模式:挂载 id 文件供 $in 过滤(bash 3.2 + set -u 下空数组要用 ${arr[@]+...} 展开)
+MONGO_ARGS=()
+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));
+  const q = { status: { $in: [3, 4] } };
+  if (process.env.COHORT === "1") {
+    // 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) };
+  }
   let first = true;
   print("[");
-  db.med_emr_info.find({ status: { $in: [3, 4] } }).forEach((d) => {
+  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),

apps/pac-service/src/cli/refresh-clinic-names.cli.ts

@@ -11,7 +11,9 @@
  * 用法:
  *   pnpm refresh-clinic-names -- --dir=./data/jvs-dw [--host=<name 覆盖>]
  *
- * 幂等:整表 upsert host.clinicNames(全量覆盖为最新派生结果)。可接 cron 定期刷。
+ * 幂等:派生结果**合并**进 host.clinicNames(同 key 新值胜,旧 key 保留)。可接 cron 定期刷。
+ *   合并而非覆盖:文件模式的导出可能是 cohort 子集(export.sh --clinics/--since),
+ *   覆盖会把子集外诊所的名字刷掉。
  */
 import { NestFactory } from '@nestjs/core';
 import { Logger } from '@nestjs/common';
@@ -121,8 +123,14 @@ async function main(): Promise<void> {
     const prisma = app.get(PrismaService);
     const host = await prisma.host.findFirst({ where: { name: hostName } });
     if (!host) throw new Error(`host "${hostName}" 不存在`);
-    await prisma.host.update({ where: { id: host.id }, data: { clinicNames: clinicMap } });
-    logger.log(`✓ host="${hostName}" clinicNames ${nc} 家(样例: ${Object.values(clinicMap).slice(0, 3).join(' / ')})`);
+    // 合并而非整表覆盖:导出文件可能是 cohort 子集(export.sh --clinics/--since),
+    // 覆盖会把子集外诊所的名字刷掉。同 key 新值胜(改名照常生效),旧 key 保留。
+    const merged = { ...((host.clinicNames as Record<string, string>) ?? {}), ...clinicMap };
+    await prisma.host.update({ where: { id: host.id }, data: { clinicNames: merged } });
+    logger.log(
+      `✓ host="${hostName}" clinicNames 本次派生 ${nc} 家,合并后 ${Object.keys(merged).length} 家` +
+        `(样例: ${Object.values(clinicMap).slice(0, 3).join(' / ')})`,
+    );
   } finally {
     await app.close();
   }

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

@@ -903,6 +903,15 @@ export class ColdImportService {
         }
       } else {
         // ── single-shot 模式(向后兼容 / 文件源)──
+        // --clinics/--since 挂在 sql_source.cohort(PAC 生成 WHERE);文件源等价物在导出侧
+        // (如 data/friday/export.sh --clinics/--since,同语义:只收窄患者名单,选中者全史全摄)。
+        // 这里显式警告而非静默忽略 —— 否则误以为过滤生效,实际全量导入。
+        if (options.clinics?.length || options.since) {
+          this.logger.warn(
+            `--clinics/--since 仅 sql_source.cohort(ClickHouse 直连)支持,当前 manifest 是文件源 → 忽略。` +
+              `文件源请在导出侧过滤(export.sh --clinics/--since),CSV 落地即已收窄。`,
+          );
+        }
         await this.processCohort({
           absDir,
           manifest,