场外核对：桥接"单据账户标识 → 平台交易账号"；保存识别字段后自动重跑核对；运营角色补推介材料权限

## 一、根因：核对查不到不是"缺数据"，是**账户口径没桥接** 实测（客户 10002 的单据）： ``` 生成的 SQL： JOIN fin_holding h ... WHERE h.trade_account = '10002' ← 单据上的账户标识结果： 0 行 → 「申请前持有份额」= null → 规则判「无法判断」→ 单据状态 query_failed ``` 而库里 `fin_holding.trade_account` 存的是**平台交易账号** `FSA{客户号:06d}` （与 `fin_sim_account.account_no` 同值，见 `tools/seed_sim_account_demo.py:226`），客户 10002 **一直持有 15911 共 1000 份**。所以页面上"单据核对与运营动作"是空的，看起来像缺数据，实为**单据印的账户标识与平台账号是两个口径，中间少了一步换算**。修法：新增 `OffsiteFundService._platform_account()`，在**单据字段落库的两个入口** （`_create_document` 新建、`_apply_recognition_fields` 保存/重试）做换算，规则： 1. 已是库里的 `account_no` → 原样； 2. 纯数字且命中 `customer_id` → 用该客户的 `account_no`； 3. 其余原样返回并留 warning —— **失败关闭，不猜不造**（不凭空生成 `FSA099999`）。一处修好，后端查询、NL2SQL 页面的自然语言、页面展示三处口径一致。附件上的 OCR **原始值不受影响**（`extracted_fields` 仍是单据上印的 `10002`）。端到端证据（同一张单据，只走服务层）： | 阶段 | document.account_identifier | 规则数据 | 单据状态 | |---|---|---|---| | 修复前 | `10002` | `{}` → 无法判断 | `query_failed` | | 保存识别字段后 | **`FSA010002`** | — | `query_failed` | | 触发核对后 | `FSA010002` | **申请前持有份额 1000.0000** → **正常** | **planned** | ## 二、前端：保存识别字段后自动重跑核对 `employee-operations/offsite/offsite.js`：原先 `save-ocr` 只保存 + 重载页面， **不触发核对**，于是运营改完字段点保存，那两个区块要么停留在上一次核对的状态、要么整块是空的，必须再手动点一次"重新核对并判定规则"——看起来像"保存没生效"。现在：保存 == 运营已人工确认该单据内容，因此保存成功后**自动对该附件关联的单据** 执行「触发 NL2SQL → 拉取返回字段 → 重新判定规则」，并在提示里区分"已重新核对" 与"核对未全部成功"。顺带把这段逻辑抽成 `recalculateDocument(taskId)`，与面板上的"重新核对并判定规则" 按钮**走同一条路径** —— 两处各写一份正是"保存后不刷新"这类不一致的来源。 ## 三、运营（operator）角色权限 `tools/grant_operator_role.py` 的授权清单补齐：`promotion:write/read/review/deliver` （`promotion_material_service.py` 的八处 `_require` 恰好只用这四个码，缺任一都会 403，例如只给 write 会在查看详情 read 那一步被拒）+ `agent:run`。已实际执行并**从身份侧验证**（`IdentityRepository.load_context`）： 9005 / 9006 现在各 7 项权限，四个推介材料码齐全。 - NL2SQL 全库与它相关的权限码**只有 `financial:nl2sql:read`**（`offsite:nl2sql` 在 `sys_permission` 里并不存在，是 `offsite_fund_service` 里 any-of 校验的死值），该码运营早已有，本次无需新增。 - 可持续性已核实：`sys_role_permission` / `sys_user_role` **没有外键**，重跑 `seed_test_rbac.py`（DELETE 重建 9001-9099 号段权限）**不会**删掉运营的绑定；且本工具按**权限码查 id**、不写死 id，天然抗号段变动。 ## 四、10001 / 10002 的 15911 持仓复核结论：**各 1000 份，且三处口径一致**（`fin_holding.market_value` = 数量 × 最新净值、净值历史 120 条、`fin_product.current_nav` 与净值最新一条一致、账户可用资金正常）。 `fin_holding` 的唯一键是 `(customer_id, product_id)`，所以**不能**再插一行 `trade_account='10002'` 的"同一个持仓"—— 那会把持仓重复计数，是错的。需要改数量就用 `python tools/seed_custom_holdings.py --quantity N`（默认就是这两个客户 + 15911）。 ## 五、验证与回归 - 新增 `tests/unit/service/test_offsite_account_bridge.py`（5 条：账号原样 / 客户号换算 / 认不出原样返回 / 不凭空造账号 / 空值不查库）； - `pytest tests/unit/service -k offsite` → 34 passed； - 场外集成测试 4 个文件 → 27 passed； - `ruff` 干净；`mypy app` 仍只有组员新代码里那 3 个既有错（与本次无关）； - 前端 `node --check offsite.js` 通过。
2026-09-14 22:16:29 +08:00
parent 292acd2e2c
commit 1d6c32f6b3
4 changed files with 228 additions and 40 deletions
@@ -1,6 +1,7 @@
 """场外基金申购赎回业务编排服务。"""

 import asyncio
+import logging
 import os
 import stat
 from collections import defaultdict
@@ -32,6 +33,7 @@ from app.core.offsite_fund_contracts import (
    RecognizedAttachment,
 )
 from app.model.audit import InteractionAudit
+from app.model.fund import FundSimAccount
 from app.model.offsite_fund import (
    OffsiteExecutionPlanTask,
    OffsiteFieldCorrection,
@@ -67,6 +69,8 @@ from app.service.offsite_smtp_adapter import (

 SHANGHAI = ZoneInfo("Asia/Shanghai")

+logger = logging.getLogger(__name__)
+
 # 浏览器可以安全内联渲染的附件类型；其余类型一律走下载，避免渲染失败或引入 XSS。
 INLINE_PREVIEW_MEDIA_TYPES = {
    "application/pdf": "application/pdf",
@@ -956,7 +960,7 @@ class OffsiteFundService:
                    self._merge_corrections(attachment.extracted_fields, cleaned)
                )
                for document in documents_by_attachment.get(attachment_id, []):
-                    self._apply_recognition_fields(document, effective)
+                    await self._apply_recognition_fields(document, effective)
                    document.updated_at = now
                saved.append(attachment_id)

@@ -1752,7 +1756,7 @@ class OffsiteFundService:
                    self._recognized_attachment_for_retry(source, result),
                    cast(Literal["subscription", "redemption"], document.document_type),
                )
-                self._apply_recognition_fields(document, fields)
+                await self._apply_recognition_fields(document, fields)
                document.operator_decision = "未处理"
                document.status = status
                document.updated_at = finished_at
@@ -1891,15 +1895,70 @@ class OffsiteFundService:
                task.status = "待执行"
            task.updated_at = now

-    @staticmethod
-    def _apply_recognition_fields(
-        document: OffsiteFundDocument, fields: Mapping[str, object]
+    async def _platform_account(self, raw: str | None) -> str | None:
+        """把单据上的「账户标识」换算成本平台的**交易账号**（`fin_sim_account.account_no`）。
+
+        ## 为什么需要这一步（2026-09-14 实测的"核对查不到"根因）
+
+        场外核对的 NL2SQL 查询计划按 `fin_holding.trade_account = <账户标识>` 过滤
+        （`nl2sql_yc.py:509-514`），而本平台 `fin_holding.trade_account` 存的是**交易账号**
+        `FSA{客户号:06d}`（与 `fin_sim_account.account_no` 同值，见
+        `tools/seed_sim_account_demo.py:226`）。单据（申购单扫描件）上印的却是投资者的
+        **账户标识**（实测 `10002`）—— 两者不同值，于是：
+
+        ```
+        JOIN fin_holding h ... WHERE h.trade_account = '10002'  → 0 行
+        → 「申请前持有份额」= null → 规则判「无法判断」→ 单据状态 query_failed
+        ```
+
+        而持仓**一直都在库里**（客户 10002 持有 15911 共 1000 份）。所以这不是"缺数据"，
+        是**账户口径没有桥接**：单据说什么，查询就按什么去比，中间少了一步换算。
+
+        ## 换算规则（逐级回退；换不出来就按原值查 —— 失败关闭，不猜不造）
+
+        1. 已经是库里的 `account_no` → 原样使用；
+        2. 纯数字且命中某个 `customer_id` → 用该客户的 `account_no`；
+        3. 其余原样返回并留 warning：宁可继续查不到（暴露给人工），也不构造一个
+           "看起来对"的账号。
+
+        单据字段是**标准化展示值**；附件上的 OCR 原始值（`extracted_fields`）不受影响，
+        仍能看到单据上原本印的是什么。
+        """
+        if not raw:
+            return None
+        existing = await self.session.scalar(
+            select(FundSimAccount.account_no).where(FundSimAccount.account_no == raw)
+        )
+        if existing is not None:
+            return str(existing)
+        if raw.isdigit():
+            account_no = await self.session.scalar(
+                select(FundSimAccount.account_no).where(
+                    FundSimAccount.customer_id == int(raw)
+                )
+            )
+            if account_no is not None:
+                logger.info(
+                    "场外核对账户换算：单据账户标识 %s → 平台交易账号 %s", raw, account_no
+                )
+                return str(account_no)
+        logger.warning(
+            "场外核对拿到无法换算的账户标识 raw=%r：按原值查询（预计查不到持仓，"
+            "可让运营在识别字段里修正后重新保存）",
+            raw,
+        )
+        return raw
+
+    async def _apply_recognition_fields(
+        self, document: OffsiteFundDocument, fields: Mapping[str, object]
    ) -> None:
        fields = normalize_recognition_fields(fields)
        raw_date = fields.get("申请日期")
        document.fund_code = OffsiteFundService._text(fields.get("基金代码"))
        document.fund_name = OffsiteFundService._text(fields.get("基金名称"))
-        document.account_identifier = OffsiteFundService._text(fields.get("账户标识"))
+        document.account_identifier = await self._platform_account(
+            OffsiteFundService._text(fields.get("账户标识"))
+        )
        document.investor_name = OffsiteFundService._text(
            fields.get("投资者名称") or fields.get("客户标识")
        )
@@ -2688,7 +2747,9 @@ class OffsiteFundService:
            task_id=task_id, mail_id=mail_id, attachment_id=attachment_id,
            document_type=document_type, fund_code=self._text(fields.get("基金代码")),
            fund_name=self._text(fields.get("基金名称")),
-            account_identifier=self._text(fields.get("账户标识")),
+            account_identifier=await self._platform_account(
+                self._text(fields.get("账户标识"))
+            ),
            investor_name=self._text(fields.get("投资者名称") or fields.get("客户标识")),
            application_no=self._text(fields.get("申请编号")),
            application_date=parse_application_date(raw_date),
@@ -471,6 +471,44 @@ if (requireOperator()) {
    window.setTimeout(() => URL.revokeObjectURL(url), 60000);
  }

+  /**
+   * 重新核对单个单据：触发 NL2SQL → 拉取返回字段 → 重新判定规则。
+   * 返回是否**全部成功**（`query_failed` 视为未成功，菜单里会提示，但不抛错）。
+   *
+   * 抽成函数是因为它有两个入口：面板上的"重新核对并判定规则"按钮，
+   * 以及保存 OCR 识别字段之后（见 `save-ocr:`）—— 两处必须走同一条路径，
+   * 否则"保存后字段不显示"这类不一致会再次出现。
+   */
+  async function recalculateDocument(taskId) {
+    if (!taskId || state.recalculatingTasks.has(taskId)) return true;
+    state.recalculatingTasks.add(taskId);
+    render();
+    try {
+      const triggerResponse = await apiClient.post(
+        'OFFSITE_TRIGGER_NL2SQL',
+        { operator_id: operatorId, manual_confirmed: true },
+        { pathParams: { taskId } },
+      );
+      const [fieldsResponse, rulesResponse] = await Promise.all([
+        apiClient.get('OFFSITE_NL2SQL_FIELDS', { pathParams: { taskId } }),
+        apiClient.post(
+          'OFFSITE_RULE_RECALCULATE',
+          { operator_id: operatorId },
+          { pathParams: { taskId } },
+        ),
+      ]);
+      state.nl2sql[taskId] = fieldsResponse.data;
+      state.nlDrafts[taskId] = {
+        ...(fieldsResponse.data?.effective_fields || fieldsResponse.data?.fields || {}),
+      };
+      state.rules[taskId] = rulesResponse.data;
+      return triggerResponse.data?.status !== 'query_failed';
+    } finally {
+      state.recalculatingTasks.delete(taskId);
+      render();
+    }
+  }
+
  async function run(action) {
    try {
      if (action === 'refresh') { await load(); showToast('邮件列表已刷新'); return; }
@@ -495,7 +533,28 @@ if (requireOperator()) {
        state.ocrDrafts[attachmentId] = { ...(saved?.effective_fields || {}) };
        syncDocumentsFromRecognition(state.recognition);
        await loadMail(state.selectedMailId);
-        showToast('OCR 识别字段修正已保存'); return;
+        // ★ 保存识别字段 == 运营已人工确认这张单据的内容，因此顺手把**该附件关联的单据**
+        //   重新核对一次（触发 NL2SQL → 拉字段 → 重新判定规则）。
+        //
+        //   为什么必须在这里做：核对结果（"单据核对与运营动作"里的 NL2SQL 返回字段与规则结果）
+        //   只在触发核对时才生成；保存识别字段本身不触发它。少了这一步，运营改完字段点保存后，
+        //   那两个区块要么停留在**上一次**核对的状态（例如仍是"申请前持有份额 null / 无法判断"），
+        //   要么整块是空的，必须再手动点一次"重新核对并判定规则"—— 而运营会以为"保存没生效"。
+        //   只核对本附件关联的单据，不动同一封邮件里的其它单据。
+        const recalcTasks = (state.documents || [])
+          .filter((document) => document.attachment_id === attachmentId)
+          .map((document) => document.task_id)
+          .filter(Boolean);
+        let recalcFailed = 0;
+        for (const taskId of recalcTasks) {
+          if (!await recalculateDocument(taskId)) recalcFailed += 1;
+        }
+        await loadMail(state.selectedMailId);
+        showToast(
+          recalcFailed ? 'OCR 识别字段已保存；核对未全部成功' : 'OCR 识别字段已保存并已重新核对',
+          recalcFailed ? 'error' : 'success',
+        );
+        return;
      }
      if (action.startsWith('save-nl:')) {
        const taskId = decodeURIComponent(action.slice(8));
@@ -515,37 +574,9 @@ if (requireOperator()) {
      if (action.startsWith('recalculate:')) {
        const taskId = decodeURIComponent(action.slice(12));
        if (state.recalculatingTasks.has(taskId)) return;
-        state.recalculatingTasks.add(taskId);
-        render();
-        try {
-          const triggerResponse = await apiClient.post(
-            'OFFSITE_TRIGGER_NL2SQL',
-            { operator_id: operatorId, manual_confirmed: true },
-            { pathParams: { taskId } },
-          );
-          const [fieldsResponse, rulesResponse] = await Promise.all([
-            apiClient.get('OFFSITE_NL2SQL_FIELDS', { pathParams: { taskId } }),
-            apiClient.post(
-              'OFFSITE_RULE_RECALCULATE',
-              { operator_id: operatorId },
-              { pathParams: { taskId } },
-            ),
-          ]);
-          state.nl2sql[taskId] = fieldsResponse.data;
-          state.nlDrafts[taskId] = {
-            ...(fieldsResponse.data?.effective_fields || fieldsResponse.data?.fields || {}),
-          };
-          state.rules[taskId] = rulesResponse.data;
-          showToast(
-            triggerResponse.data?.status === 'query_failed'
-              ? '已重新核对，但存在查询失败'
-              : '规则已重新核对并判定',
-            triggerResponse.data?.status === 'query_failed' ? 'error' : 'success',
-          );
-        } finally {
-          state.recalculatingTasks.delete(taskId);
-          render();
-        }
+        const ok = await recalculateDocument(taskId);
+        showToast(ok ? '规则已重新核对并判定' : '已重新核对，但存在查询失败',
+                  ok ? 'success' : 'error');
        return;
      }
      if (action.startsWith('confirm:')) {
@@ -0,0 +1,85 @@
+"""「单据账户标识 → 平台交易账号」换算（`OffsiteFundService._platform_account`）的回归测试。
+
+## 为什么必须守住这条缝
+
+场外核对的 NL2SQL 查询计划按 `fin_holding.trade_account = <单据上的账户标识>` 过滤
+（`nl2sql_yc.py:509-514`），而库里存的是平台交易账号 `FSA{客户号:06d}`。两者**不同值**时：
+
+```
+JOIN fin_holding h ... WHERE h.trade_account = '10002'  → 0 行
+→ 「申请前持有份额」= null → 规则判「无法判断」→ 单据状态 query_failed
+```
+
+2026-09-14 实测就是这样：客户 10002 **明明持有 15911 共 1000 份**，核对却报"查询无可用数据"，
+页面上"单据核对与运营动作"因此是空的 —— 看起来像"缺数据"，实为**账户口径没桥接**。
+本文件把换算规则钉死：命中就换算，认不出来就**原样返回（失败关闭，不猜不造）**。
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from app.service.offsite_fund_service import OffsiteFundService
+
+
+class _StubSession:
+    """按调用顺序返回预置结果的 session 替身；记录被执行的查询，不发真实 SQL。"""
+
+    def __init__(self, results: list[Any]) -> None:
+        self._results = list(results)
+        self.queries: list[str] = []
+
+    async def scalar(self, statement: Any) -> Any:
+        self.queries.append(str(statement))
+        return self._results.pop(0) if self._results else None
+
+
+def _service(results: list[Any]) -> tuple[OffsiteFundService, _StubSession]:
+    session = _StubSession(results)
+    return OffsiteFundService(session), session  # type: ignore[arg-type]
+
+
+@pytest.mark.asyncio
+async def test_account_no_is_used_as_is() -> None:
+    """已经是库里的交易账号：原样使用，且**只查一次**（不需要按客户号回退）。"""
+    service, session = _service(["FSA010002"])
+
+    assert await service._platform_account("FSA010002") == "FSA010002"
+    assert len(session.queries) == 1
+
+
+@pytest.mark.asyncio
+async def test_numeric_customer_id_is_translated_to_trade_account() -> None:
+    """单据上印的是客户号（演示单据就是 `10002`）⇒ 换算成该客户的交易账号。"""
+    service, session = _service([None, "FSA010002"])
+
+    assert await service._platform_account("10002") == "FSA010002"
+    assert len(session.queries) == 2  # 先按 account_no 找、再按 customer_id 找
+
+
+@pytest.mark.asyncio
+async def test_unknown_identifier_is_returned_unchanged() -> None:
+    """认不出来就原样返回：宁可继续查不到（暴露给人工），也不构造"看起来对"的账号。"""
+    service, _session = _service([None, None])
+
+    assert await service._platform_account("SL-UNKNOWN-001") == "SL-UNKNOWN-001"
+
+
+@pytest.mark.asyncio
+async def test_numeric_identifier_without_account_is_not_invented() -> None:
+    """数字但是个不存在的客户号 ⇒ 不得凭空造出 `FSA099999` 这类账号。"""
+    service, _session = _service([None, None])
+
+    assert await service._platform_account("99999") == "99999"
+
+
+@pytest.mark.asyncio
+async def test_blank_identifier_needs_no_query() -> None:
+    """空值不查库。"""
+    service, session = _service([])
+
+    assert await service._platform_account(None) is None
+    assert await service._platform_account("") is None
+    assert session.queries == []
@@ -56,8 +56,19 @@ OPERATOR_GRANTED_CODES: tuple[str, ...] = (
    "agent:run",
    # 场外运营（角色门槛之外，这个码是场外线自己声明的）
    "offsite:write",
-    # 金融 NL2SQL：角色白名单 {advisor, operator, admin, super_admin} 含 operator
+    # 金融 NL2SQL：角色白名单 {advisor, operator, admin, super_admin} 含 operator。
+    # ⚠️ 全库与 NL2SQL 相关的权限码**只有这一个**（`offsite:nl2sql` 在
+    # `sys_permission` 里并不存在，是 `offsite_fund_service` 里 any-of 校验的死值），
+    # 所以"NL2SQL 的全部权限"就是它。
    "financial:nl2sql:read",
+    # 产品推介材料（生成/查看/审核/发送）：`promotion_material_service.py` 的八处
+    # `_require` 恰好只用到这四个码（72/91/112/136 要 write，150/161 要 read，
+    # 211 要 review，229 要 deliver）。运营要"生成推介材料"就必须四个齐全 ——
+    # 只给 write 会在查看详情（read）那一步 403。
+    "promotion:write",
+    "promotion:read",
+    "promotion:review",
+    "promotion:deliver",
 )