fix(摄入): 影像 AI 的诊所取自这张片自己那次病历,⛔ 不再"借"患者别处的诊所

`image_finding_rows` 原来用 `any(organization_id)` 按**患者**取诊所:随便挑该患者的
一条病历,把那家诊所贴到片子上。两个毛病:
  ① `any()` 语义就是「随便挑一条」—— 患者跨诊所时挑中哪家是任意的,重跑还会变;
  ② 挑的维度根本不对 —— 要回答的是「**这张片在哪拍的**」,却按「这个人在哪看过」作答。
影像表自己没有诊所列(19 个列全查过),但**有 `emr_id` 且 100% 有值**,指向这张片所属的
那次病历 —— 真来源一直就在手边。

⇒ 改成按 `ia.emr_id` 精确 join 那条病历取 `organization_id`,并加 `e.patient_id = c.patient_id`
  (片子的病历必须属于同一个患者)。

📊 全量实测(2026-08-16,测试服连的是真 DW):
  · 两版都能算出诊所的 160,436 行里,**24,849 行(15.5%)现在挂错**,牵涉 21,149 个患者。
  · 改后只少 879 行(0.5%)—— 那些片子挂着一个**在病历表里不存在**的 emr_id。
🔴 少掉的那些是**故意不要**的(产品定):取不到诊所 → `transaction-synthesizer` 按
  「clinic 是立柱必填」跳过 → 不摄入。⛔ 宁可不要,也不拿别的诊所顶上去装作知道。

🔍 抓到这条的现场:测试服上海世纪公园池子里三个"算不出档位"的患者,病历号前缀是
  **CQ / GZ**(重庆 / 广州),被贴成了上海 —— 而那家诊所在池患者 99.3% 是 SH 开头。
  顺着查:他们片子的 emr_id 在病历表里查不到,当年借来的那个诊所**今天也借不到了**
  (`po.org` 现在返回 NULL)⇒ 那三条事实的诊所是个**无法复现、无法验证**的值。

⚠️ `GROUP BY` 里 `patient_id` 必须留着:cohort 注入是插到**首个 GROUP BY 前**的,
  改成只按 id 分组会让注入落空 → 每批全表扫 + 串批,而且不报错。
  已按 `injectCohortFilter` 的真实行为(已有 WHERE 时追加 `AND`)在 DW 上验过注入后可执行。

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

apps/pac-service/data/jvs-dw/manifest.yaml

@@ -248,8 +248,23 @@ sql_source:
     #   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 验证等价正确)。
+    # ⚠️ cohort 注入(裸 patient_id,brand IN)会插到**首个 GROUP BY 前** = 下方 e 子查询 →
+    #   e 按 batch scope,INNER JOIN 恰好把输出过滤到 batch 患者。
+    #   🔴 所以那个 GROUP BY 里 **`patient_id` 必须留着**,⛔ 别"优化"成只按 id 分组 ——
+    #   注入会落空 → 每批全表扫 + 串批,而且**不报错**。
+    #
+    # 🔴 **诊所取自这张片自己那次病历(`ia.emr_id`),⛔ 不许再"借"**(2026-08-16 改)。
+    #   原来是 `any(organization_id)` 按**患者**取:随便挑该患者的一条病历,把那家诊所
+    #   贴到片子上。两个毛病 ——
+    #     ① `any()` 语义就是"随便挑一条",患者跨诊所时挑中哪家是任意的、重跑还会变;
+    #     ② 挑的维度根本不对:要答的是「**这张片在哪拍的**」,却按"这个人在哪看过"回答。
+    #   实测(2026-08-16 全量):160,436 条两版都能算出诊所的行里,**24,849 条(15.5%)挂错**,
+    #   牵涉 21,149 个患者。测试服上海世纪公园抓到三个样本,病历号前缀是 CQ/GZ(重庆/广州),
+    #   被贴成了上海 —— 而那家诊所在池患者 99.3% 是 SH 开头的。
+    #   ⚠️ 改成精确 join 只少 879 行(0.5%):那些是片子挂着一个**在病历表里不存在**的 emr_id。
+    #     它们取不到诊所 → `transaction-synthesizer` 按「clinic 是立柱必填」跳过 → 不摄入。
+    #     ⭐ 这是刻意的(产品定):宁可不要,也⛔ 不拿别的诊所顶上去装作知道。
+    #   ⚠️ `e.patient_id = c.patient_id` 这一条也要留:片子的病历必须属于同一个患者。
     # 病种→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,
@@ -261,7 +276,7 @@ sql_source:
                -- 注:同 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,
+          SELECT c.patient_id AS patient_id, c.brand AS brand, e.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,
@@ -269,12 +284,12 @@ sql_source:
                  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
+          INNER JOIN (
+            SELECT patient_id, brand, id, 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)
+            GROUP BY patient_id, brand, id
+          ) e ON e.id = ia.emr_id AND e.brand = ia.brand AND e.patient_id = c.patient_id
+          WHERE c.last_visit_time IS NOT NULL
         ) s1
         ARRAY JOIN [('K02', cavity), ('K01', impacted_tooth), ('K01', embedded_tooth),
                     ('K04', root_periodontitis), ('K03', root_remnant), ('K03', crown_remnant),