refactor#卡片样式调整

code/post_clear_check/sql/schema.sql

@@ -48,15 +48,15 @@ CREATE TABLE check_run (
   )
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='一轮核对';
 
--- 一张飞书卡片对应一行。明细卡是一条链路、一个归属对象。
--- 概况卡 card_kind 为 chain_summary，一条链路一行，biz_key 写链路代码，没有 check_line。
+-- 一张飞书卡片对应一行。核对完成后对外是四张方向卡，card_kind 为 dimension。
+-- 旧的归属卡和概况卡仍允许留在历史行里。门禁没开跑时仍用 clearing_batch。
 -- 同一 dedupe_key 已是 sent 则不再发送。failed 只重发这一条，不新插一行。
--- chain_name 存链路英文代码，卡片上显示中文名。空串只允许回滚后的旧代码写入，新代码不写空串。
+-- chain_name 存方向代码或旧的内部链路代码，卡片上显示中文名。空串只允许回滚后的旧代码写入。
 CREATE TABLE check_result (
   id BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键，卡片正文里的记录编号',
   run_id BIGINT NOT NULL COMMENT 'check_run.id',
-  card_kind VARCHAR(32) NOT NULL COMMENT '归属种类：master_order、fund_account、trs_user、clearing_batch；chain_summary 为链路概况',
-  chain_name VARCHAR(32) NOT NULL DEFAULT '' COMMENT '业务链路代码，见业务链路核对设计 4.1；unassigned 为未归链路',
+  card_kind VARCHAR(32) NOT NULL COMMENT 'dimension 为方向卡；历史行仍有 master_order、fund_account、trs_user、clearing_batch、chain_summary',
+  chain_name VARCHAR(32) NOT NULL DEFAULT '' COMMENT '方向代码 order_chain、issue_report、sales_report、trs_report；历史行仍是内部链路代码',
   biz_key VARCHAR(64) NOT NULL COMMENT '主订单编号、资金账户、TRS 用户账号、批次号或营业日；概况卡为链路代码',
   verdict VARCHAR(16) NOT NULL COMMENT 'correct、error、suspect',
   dedupe_key VARCHAR(191) NOT NULL COMMENT '交易日、卡片种类、链路代码、业务键、三端批次、三端 end_time、规则版本',
@@ -67,11 +67,17 @@ CREATE TABLE check_result (
   UNIQUE KEY uk_check_result_dedupe (dedupe_key),
   KEY idx_check_result_run_id (run_id, notify_status),
   CONSTRAINT chk_check_result_card_kind CHECK (
-    card_kind COLLATE utf8mb4_bin IN ('master_order', 'fund_account', 'trs_user', 'clearing_batch', 'chain_summary')
+    card_kind COLLATE utf8mb4_bin IN (
+      'dimension', 'master_order', 'fund_account', 'trs_user', 'clearing_batch', 'chain_summary'
+    )
   ),
   CONSTRAINT chk_check_result_chain_name CHECK (
     chain_name COLLATE utf8mb4_bin IN (
       '',
+      'order_chain',
+      'issue_report',
+      'sales_report',
+      'trs_report',
       'buy_execution',
       'sell_execution',
       'cash_change',

code/post_clear_check/src/post_clear_check/chain.py

@@ -86,6 +86,39 @@ CHAIN_GATE = "check_gate"
 # 收容桶，不是业务链路。挂不上链路或没有归属键的行进这里，结论一律记错误。
 CHAIN_UNASSIGNED = "unassigned"
 
+# 对外只发这四个方向。card_kind 用 dimension，chain_name 用下面的代码。
+# 结论行上的内部链路代码不改，方向只决定收进哪一张卡。
+CARD_DIMENSION = "dimension"
+DIM_ORDER = "order_chain"
+DIM_ISSUE = "issue_report"
+DIM_SALES = "sales_report"
+DIM_TRS = "trs_report"
+DIMENSIONS = (DIM_ORDER, DIM_ISSUE, DIM_SALES, DIM_TRS)
+DIMENSION_NAME = {
+    DIM_ORDER: "订单链路",
+    DIM_ISSUE: "发行报表",
+    DIM_SALES: "销售报表",
+    DIM_TRS: "TRS报表",
+}
+# 订单卡上的索引和批次状态。不参与总结论，读回后仍靠这个前缀识别。
+INDEX_MARK = "关联报表索引："
+STATUS_MARK = "批次状态："
+# 内部链路可以出现在哪些方向上。收卡按产生来源，这张表只用来确认没有链路被漏掉。
+CHAIN_DIMENSIONS = {
+    CHAIN_BUY: frozenset({DIM_ORDER, DIM_ISSUE, DIM_SALES}),
+    CHAIN_SELL: frozenset({DIM_ORDER, DIM_ISSUE, DIM_SALES}),
+    CHAIN_CASH_CHANGE: frozenset({DIM_SALES}),
+    CHAIN_CASH_SETTLEMENT: frozenset({DIM_ISSUE, DIM_SALES}),
+    CHAIN_ACCOUNT_ASSET: frozenset({DIM_SALES}),
+    CHAIN_NOTE_HOLDING: frozenset({DIM_ISSUE, DIM_SALES}),
+    CHAIN_TOPUP: frozenset({DIM_ISSUE, DIM_SALES}),
+    CHAIN_EXTENSION: frozenset({DIM_ISSUE, DIM_SALES}),
+    CHAIN_TRADE: frozenset({DIM_TRS}),
+    CHAIN_USER_ASSET: frozenset({DIM_TRS}),
+    CHAIN_GATE: frozenset({DIM_ORDER, DIM_ISSUE, DIM_SALES, DIM_TRS}),
+    CHAIN_UNASSIGNED: frozenset(DIMENSIONS),
+}
+
 # 报表所在的端。决定报表编号查哪一端的原型路径。
 SIDE_ISSUE = "issue"
 SIDE_SALES = "sales"

code/post_clear_check/src/post_clear_check/dimension_body.py

@@ -0,0 +1,558 @@
+"""四个方向的飞书正文。
+
+不重新计算结论，只把已经落在这张卡上的行按巡检章节排出来。
+正确项按字段或「报表 + 依据」收成一块，错误和可疑逐条展开。
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from datetime import date, datetime
+from decimal import Decimal
+
+from post_clear_check.chain import (
+    DIM_ISSUE,
+    DIM_ORDER,
+    DIM_SALES,
+    DIM_TRS,
+    DIMENSION_NAME,
+    GATE_MISSING_TABLE,
+    INDEX_MARK,
+    STATUS_MARK,
+)
+from post_clear_check.feishu import FeishuError, card_text
+from post_clear_check.gate import (
+    FORMULA_DAYCUT,
+    FORMULA_EARLY_REPORT,
+    FORMULA_ISSUE_BASELINE,
+    FORMULA_ISSUE_PUBLISH,
+    FORMULA_SALES_PUBLISH,
+    FORMULA_SALES_UPSTREAM,
+    FORMULA_TRS_BLOCKED,
+    FORMULA_TRS_DATE,
+    FORMULA_TRS_VERSION,
+)
+from post_clear_check.order_decimal import VERDICT_CORRECT, VERDICT_ERROR, VERDICT_SUSPECT
+
+# 门禁公式。读回后结论行上的链路代码会被写成方向代码，只能靠公式识别。
+GATE_FORMULAS = frozenset(
+    {
+        FORMULA_EARLY_REPORT,
+        FORMULA_TRS_BLOCKED,
+        FORMULA_ISSUE_BASELINE,
+        FORMULA_ISSUE_PUBLISH,
+        FORMULA_SALES_UPSTREAM,
+        FORMULA_SALES_PUBLISH,
+        FORMULA_TRS_DATE,
+        FORMULA_TRS_VERSION,
+        FORMULA_DAYCUT,
+        GATE_MISSING_TABLE,
+    }
+)
+_ORDER_SECTIONS = (
+    ("link", "每单链路"),
+    ("intraday", "盘中公式"),
+    ("clear", "清算对照"),
+)
+_VERDICT_RANK = {VERDICT_ERROR: 0, VERDICT_SUSPECT: 1, VERDICT_CORRECT: 2}
+
+
+class _View:
+    """渲染只用到的字段。缺归属时按空串。"""
+
+    def __init__(self, line: object) -> None:
+        self.formula_text = card_text(str(getattr(line, "formula_text", "")))
+        self.calc_text = card_text(str(getattr(line, "calc_text", "")))
+        self.calc_value = getattr(line, "calc_value", None)
+        self.check_text = card_text(str(getattr(line, "check_text", "")))
+        self.check_value = getattr(line, "check_value", None)
+        self.basis = card_text(str(getattr(line, "basis", "")))
+        self.verdict = str(getattr(line, "verdict", ""))
+        self.report_code = card_text(str(getattr(line, "report_code", "") or ""))
+        self.master_order_no = card_text(str(getattr(line, "master_order_no", "") or ""))
+        self.fund_account = card_text(str(getattr(line, "fund_account", "") or ""))
+        self.user_account = card_text(str(getattr(line, "user_account", "") or ""))
+
+
+def render_dimension_body(
+    *,
+    chain_name: str,
+    business_date: date,
+    trade_date: date,
+    batch_no: str,
+    lines: Sequence[object],
+    record_no: str,
+    extra: str = "",
+) -> str:
+    """一个方向一篇。错误主题也保留正确章节。超长只删正确块。"""
+
+    if chain_name not in DIMENSION_NAME:
+        raise FeishuError("卡片方向不在四个方向之内")
+    views = tuple(_View(line) for line in lines)
+    for view in views:
+        if view.verdict not in _VERDICT_RANK:
+            raise FeishuError("结论不在正确、错误、可疑之内")
+    header = "\n".join(
+        (
+            f"营业日：{_day(business_date)}",
+            f"交易日：{_day(trade_date)}",
+            f"批次号：{_one_line(batch_no)}",
+        )
+    )
+    tail = card_text(extra).strip()
+    # 方向卡按巡检稿把每张报表、每段公式写全。不按 3500 字删章，否则整章会被收成「另有 1 项」。
+    # 请求体超过约 28KB 时由发送侧按行拆卡。
+    pieces = [header, *_body(chain_name, views, business_date)]
+    if record_no and "记录编号" not in tail:
+        pieces.append(f"记录编号：{_one_line(record_no)}")
+    if tail:
+        pieces.append(tail)
+    return "\n\n".join(piece for piece in pieces if piece)
+
+
+def _body(chain_name: str, views: tuple[_View, ...], business_date: date) -> list[str]:
+    """章节之间空一行，章节内部用单个换行。空清单收成一行。"""
+
+    annotations = [view for view in views if view.formula_text.startswith(INDEX_MARK)]
+    gates = [view for view in views if view.formula_text in GATE_FORMULAS]
+    rest = [
+        view
+        for view in views
+        if not view.formula_text.startswith(INDEX_MARK)
+        and not view.formula_text.startswith(STATUS_MARK)
+        and view.formula_text not in GATE_FORMULAS
+    ]
+    pieces = [_summary(chain_name, views)]
+    if chain_name == DIM_ORDER:
+        pieces.extend(_order_sections(rest))
+        pieces.append(_index_chapter(annotations))
+    else:
+        if gates:
+            pieces.append(_tight("批次门禁", [_gate_block(view) for view in gates]))
+        pieces.extend(_report_sections(rest, business_date))
+    errors = [view for view in rest if view.verdict == VERDICT_ERROR]
+    blocked = [view for view in rest if view.verdict == VERDICT_SUSPECT and view.calc_value is None]
+    suspects = [view for view in rest if view.verdict == VERDICT_SUSPECT and view.calc_value is not None]
+    pieces.append(_lists(errors, suspects, blocked))
+    return pieces
+
+
+def _summary(chain_name: str, views: tuple[_View, ...]) -> str:
+    """摘要按主单或报表计笔。订单卡把三端门禁收进行内，不再各占一章。"""
+
+    scored = [
+        view
+        for view in views
+        if not view.formula_text.startswith(INDEX_MARK)
+        and not view.formula_text.startswith(STATUS_MARK)
+        and view.formula_text not in GATE_FORMULAS
+    ]
+    buckets: dict[str, str] = {}
+    for view in scored:
+        if chain_name == DIM_ORDER:
+            key = view.master_order_no or view.formula_text
+        else:
+            key = view.report_code or view.formula_text
+        buckets[key] = _worse(buckets.get(key), view.verdict)
+    counts = {VERDICT_CORRECT: 0, VERDICT_ERROR: 0, VERDICT_SUSPECT: 0}
+    for verdict in buckets.values():
+        counts[verdict] += 1
+    unit = "笔" if chain_name == DIM_ORDER else "张报表"
+    rows = ["摘要"]
+    if chain_name == DIM_ORDER:
+        rows.append(
+            "人口：已完成或已结算的 ORD 主单。"
+            f"本方向正确 {counts[VERDICT_CORRECT]} {unit}，"
+            f"错误 {counts[VERDICT_ERROR]} {unit}，"
+            f"可疑 {counts[VERDICT_SUSPECT]} {unit}。"
+            "已撤主单、强平主单不进入买入/卖出公式。"
+        )
+        for view in views:
+            if view.formula_text.startswith(STATUS_MARK):
+                rows.append(view.calc_text or view.check_text)
+    else:
+        rows.append(
+            f"本方向正确 {counts[VERDICT_CORRECT]} {unit}，"
+            f"错误 {counts[VERDICT_ERROR]}，"
+            f"可疑 {counts[VERDICT_SUSPECT]}。"
+        )
+    return "\n".join(rows)
+
+
+def _order_sections(rows: Sequence[_View]) -> list[str]:
+    """每单链路只写总结。盘中公式和清算对照按字段写五行，字段之间空一行。"""
+
+    grouped: dict[str, list[_View]] = {code: [] for code, _title in _ORDER_SECTIONS}
+    for view in rows:
+        grouped[_order_section(view)].append(view)
+    found: list[str] = []
+    for code, title in _ORDER_SECTIONS:
+        section_rows = grouped[code]
+        if code == "link":
+            found.append(_tight(title, [_link_summary(section_rows)]))
+            continue
+        blocks = _field_blocks(section_rows)
+        found.append(_tight(title, blocks) if blocks else _tight(title, ["本轮没有这类结论。"]))
+    return found
+
+
+def _link_summary(rows: Sequence[_View]) -> str:
+    """每单链路不列表。按主单汇总执行事实和对冲执行根。"""
+
+    if not rows:
+        return "本轮没有执行事实或对冲执行根结论。"
+    orders: dict[str, str] = {}
+    for view in rows:
+        key = view.master_order_no or view.formula_text
+        orders[key] = _worse(orders.get(key), view.verdict)
+    counts = {VERDICT_CORRECT: 0, VERDICT_ERROR: 0, VERDICT_SUSPECT: 0}
+    for verdict in orders.values():
+        counts[verdict] += 1
+    text = (
+        f"本轮主单 {len(orders)} 笔。执行事实与对冲执行根："
+        f"正确 {counts[VERDICT_CORRECT]} 笔，"
+        f"错误 {counts[VERDICT_ERROR]} 笔，"
+        f"可疑 {counts[VERDICT_SUSPECT]} 笔。"
+    )
+    bad = [key for key, verdict in orders.items() if verdict != VERDICT_CORRECT]
+    if bad:
+        text += "未通过：" + "、".join(bad[:8])
+        if len(bad) > 8:
+            text += f"…等共 {len(bad)} 笔"
+        text += "。"
+    return text
+
+
+def _field_blocks(rows: Sequence[_View]) -> list[str]:
+    """同一字段收成五行。正确写笔数，错误和可疑逐条留在清单里，这里仍给出该字段的总结论。"""
+
+    grouped: dict[tuple[str, str], list[_View]] = {}
+    for view in rows:
+        grouped.setdefault((view.formula_text, view.basis), []).append(view)
+    blocks: list[str] = []
+    for (formula, basis), group in grouped.items():
+        correct = sum(1 for view in group if view.verdict == VERDICT_CORRECT)
+        verdict = VERDICT_CORRECT
+        for view in group:
+            verdict = _worse(verdict, view.verdict)
+        orders = _owners(group, "master_order_no")
+        shown = _brief(orders) if orders else f"{len(group)} 列"
+        sample = next((view for view in group if view.verdict != VERDICT_CORRECT), group[0])
+        actual = f"{shown} 无差异" if verdict == VERDICT_CORRECT else _actual(sample)
+        label = {VERDICT_CORRECT: "正确", VERDICT_ERROR: "错误", VERDICT_SUSPECT: "可疑"}[verdict]
+        blocks.append(
+            _lines(
+                (
+                    f"字段：{formula}",
+                    f"期望：{_expect(sample) if verdict != VERDICT_CORRECT else formula}",
+                    f"实据：{actual}",
+                    f"依据：{basis}",
+                    f"结论：{label}（{correct}/{len(group)}）",
+                )
+            )
+        )
+    return blocks
+
+
+def _report_sections(rows: Sequence[_View], business_date: date) -> list[str]:
+    """每张报表一块，六行对齐巡检稿。没有报表编号的步骤节点单独成块。"""
+
+    grouped: dict[str, list[_View]] = {}
+    for view in rows:
+        grouped.setdefault(view.report_code or view.formula_text, []).append(view)
+    blocks = [_report_block(group, business_date) for _key, group in grouped.items()]
+    if not blocks:
+        return ["分报表\n本轮没有报表结论。"]
+    return blocks
+
+
+def _report_block(rows: Sequence[_View], business_date: date) -> str:
+    """一张报表的筛选、期望、实据、依据、结论。期望列公式，不另算合计。"""
+
+    label = _report_label(rows[0])
+    scope = _scope(rows)
+    screen = f"营业日 {_day(business_date)}"
+    if scope:
+        screen = f"{screen}，{scope}"
+    verdict = VERDICT_CORRECT
+    for view in rows:
+        verdict = _worse(verdict, view.verdict)
+    names = _formula_names(rows)
+    expect = "；".join(names[:6])
+    if len(names) > 6:
+        expect += f"；其余 {len(names) - 6} 列见依据"
+    correct = sum(1 for view in rows if view.verdict == VERDICT_CORRECT)
+    bases = _unique(view.basis for view in rows if view.basis)
+    basis = "；".join(bases[:4])
+    if len(bases) > 4:
+        basis += f"；另有 {len(bases) - 4} 条依据"
+    label_text = {VERDICT_CORRECT: "正确", VERDICT_ERROR: "错误", VERDICT_SUSPECT: "可疑"}[verdict]
+    actual = f"{len(rows)} 列无差异" if verdict == VERDICT_CORRECT else f"{correct}/{len(rows)} 列无差异，其余见下方清单"
+    return _lines(
+        (
+            f"报表：{label}",
+            f"筛选：{screen}",
+            f"期望：{expect}",
+            f"实据：{actual}",
+            f"依据：{basis or '见各列登记'}",
+            f"结论：{label_text}",
+        )
+    )
+
+
+def _formula_names(rows: Sequence[_View]) -> list[str]:
+    found: list[str] = []
+    for view in rows:
+        name = view.formula_text.split("=", 1)[0].strip() or view.formula_text
+        if name not in found:
+            found.append(name)
+    return found
+
+
+def _unique(values: Sequence[str]) -> list[str]:
+    found: list[str] = []
+    for value in values:
+        if value and value not in found:
+            found.append(value)
+    return found
+
+
+def _detail(view: _View) -> str:
+    """错误、可疑、无法展开逐条写计算过程。"""
+
+    rows: list[str] = []
+    label = _report_label(view)
+    if view.report_code or label != view.formula_text:
+        rows.append(f"报表：{label}")
+    scope = _scope((view,))
+    if scope:
+        rows.append(f"筛选：{scope}")
+    rows.append(f"节点：{view.formula_text}")
+    rows.append(f"期望：{_expect(view)}")
+    rows.append(f"实据：{_actual(view)}")
+    if view.basis:
+        rows.append(f"依据：{view.basis}")
+    rows.append(_conclusion(view))
+    return _lines(rows)
+
+
+def _gate_block(view: _View) -> str:
+    return _lines(
+        (
+            f"字段：{view.formula_text}",
+            f"期望：{view.check_text or view.formula_text}",
+            f"实据：{view.calc_text or '当前缺失'}",
+            f"依据：{view.basis}",
+            _conclusion(view),
+        )
+    )
+
+
+def _index_chapter(lines: Sequence[_View]) -> str:
+    """关联报表不列表。每张报表五行：字段、期望、实据、依据、结论。"""
+
+    blocks = [
+        _index_block(view.calc_text or view.check_text)
+        for view in lines
+        if view.calc_text or view.check_text
+    ]
+    if not blocks:
+        return _tight("关联报表", ["本轮主单没有挂到报表行。"])
+    return _tight("关联报表", blocks)
+
+
+def _index_block(text: str) -> str:
+    head, _, rest = text.partition("：")
+    actual = rest.split("。", 1)[0] if rest else text
+    detail = "对应报表卡"
+    if "明细在" in text:
+        detail = "明细在" + text.split("明细在", 1)[1].strip("。")
+    verdict = "正确"
+    if "错误" in text:
+        verdict = "错误"
+    elif "可疑" in text:
+        verdict = "可疑"
+    return _lines(
+        (
+            f"字段：{head or text}",
+            "期望：本轮主单在该报表上的列与对应报表卡一致",
+            f"实据：{actual or text}",
+            f"依据：{detail}",
+            f"结论：{verdict}",
+        )
+    )
+
+
+def _lists(errors: Sequence[_View], suspects: Sequence[_View], blocked: Sequence[_View]) -> str:
+    """三张清单。没有行时各写一行，不再标题下再空一行写「无」。"""
+
+    return "\n".join(
+        (
+            _list_part("错误清单", [_detail(view) for view in errors]),
+            _list_part("可疑清单", [_detail(view) for view in suspects]),
+            _list_part("无法展开", [_detail(view) for view in blocked]),
+        )
+    )
+
+
+def _list_part(title: str, blocks: Sequence[str]) -> str:
+    if not blocks:
+        return f"{title}：无"
+    return title + "\n" + "\n\n".join(blocks)
+
+
+def _tight(title: str, blocks: Sequence[str]) -> str:
+    """章标题后直接接正文。块与块之间空一行，块内不空行。"""
+
+    kept = [block for block in blocks if block]
+    if not kept:
+        return f"{title}\n无"
+    return title + "\n" + "\n\n".join(kept)
+
+
+def _order_section(view: _View) -> str:
+    """订单卡上的一行归到哪一章。报表行已经归到三张报表卡，这里只看公式。"""
+
+    formula = view.formula_text
+    if "工作快照" in formula or "已发布结果" in formula or "当前投影" in formula:
+        return "clear"
+    if formula.startswith("执行事实") or "执行根" in formula:
+        return "link"
+    return "intraday"
+
+
+def _report_label(view: _View) -> str:
+    head = view.check_text.split(".", 1)[0].split("=", 1)[0].strip()
+    if view.report_code and view.report_code in head:
+        return head
+    if view.report_code:
+        return view.report_code
+    return head or view.formula_text
+
+
+def _scope(lines: Sequence[_View]) -> str:
+    parts: list[str] = []
+    orders = _owners(lines, "master_order_no")
+    funds = _owners(lines, "fund_account")
+    users = _owners(lines, "user_account")
+    if orders:
+        parts.append("主订单 " + _brief(orders))
+    if funds:
+        parts.append("资金账户 " + _brief(funds))
+    if users:
+        parts.append("用户账号 " + _brief(users))
+    return " · ".join(parts)
+
+
+def _owners(lines: Sequence[_View], field: str) -> list[str]:
+    found: list[str] = []
+    for view in lines:
+        value = getattr(view, field)
+        if value and value not in found:
+            found.append(value)
+    return found
+
+
+def _brief(values: Sequence[str]) -> str:
+    shown = "、".join(values[:8])
+    if len(values) > 8:
+        return f"{shown}…等共 {len(values)} 个"
+    return shown
+
+
+def _expect(view: _View) -> str:
+    if view.calc_value is None and view.verdict == VERDICT_SUSPECT:
+        return "期望值无法计算"
+    head = view.calc_text.split("；", 1)[0].strip()
+    if head and head != view.formula_text:
+        return head
+    number = _number(view.calc_value)
+    if number is not None:
+        return number
+    return view.formula_text
+
+
+def _actual(view: _View) -> str:
+    current = _number(view.check_value)
+    check = view.check_text.strip()
+    if check and current is not None and current not in check:
+        return f"{check}。当前 {current}"
+    if check:
+        return check
+    if current is not None:
+        return f"当前 {current}"
+    return "当前缺失"
+
+
+def _conclusion(view: _View) -> str:
+    expected = _number(view.calc_value)
+    current = _number(view.check_value)
+    if view.verdict == VERDICT_ERROR:
+        left = expected if expected is not None else _expect(view)
+        right = current if current is not None else "缺失"
+        return f"结论：错误。期望 {left}，当前 {right}。不符合预期。"
+    if view.verdict == VERDICT_SUSPECT and view.calc_value is None:
+        right = current if current is not None else "缺失"
+        return f"结论：可疑。期望值无法计算，当前 {right}。"
+    if view.verdict == VERDICT_SUSPECT:
+        left = expected if expected is not None else _expect(view)
+        right = current if current is not None else "缺失"
+        return f"结论：可疑。期望 {left}，当前 {right}。无法判定。"
+    return "结论：正确"
+
+
+def _worse(current: str | None, verdict: str) -> str:
+    if current is None:
+        return verdict
+    if _VERDICT_RANK[verdict] < _VERDICT_RANK[current]:
+        return verdict
+    return current
+
+
+def _lines(rows: Sequence[str]) -> str:
+    """巡检稿里的六行用单个换行。飞书里再空一行会把一张报表撑散。"""
+
+    return "\n".join(row for row in rows if row)
+
+
+def _number(value: object) -> str | None:
+    if value is None:
+        return None
+    if isinstance(value, Decimal):
+        text = format(value, "f")
+        if "." in text:
+            text = text.rstrip("0").rstrip(".")
+        return text or "0"
+    return str(value)
+
+
+def _day(value: date) -> str:
+    if isinstance(value, datetime):
+        return value.date().isoformat()
+    if isinstance(value, date):
+        return value.isoformat()
+    raise FeishuError("营业日或交易日不是日期")
+
+
+def _one_line(value: str) -> str:
+    text = card_text(value).strip()
+    if not text or "\n" in text:
+        raise FeishuError("批次号或记录编号不能为空或换行")
+    return text
+
+
+def dimension_of_gate(formula: str, check_text: str) -> str:
+    """已完成轮次上的门禁行归到哪一张报表卡。订单卡只引用，不重复这几行。"""
+
+    if formula in {FORMULA_ISSUE_BASELINE, FORMULA_ISSUE_PUBLISH}:
+        return DIM_ISSUE
+    if formula in {FORMULA_SALES_UPSTREAM, FORMULA_SALES_PUBLISH, FORMULA_DAYCUT}:
+        return DIM_SALES
+    if formula in {FORMULA_TRS_DATE, FORMULA_TRS_VERSION, FORMULA_TRS_BLOCKED}:
+        return DIM_TRS
+    text = check_text
+    if "db_trs" in text or "，TRS，" in text or text.startswith("TRS"):
+        return DIM_TRS
+    if "db_sales" in text or "销售" in text:
+        return DIM_SALES
+    return DIM_ISSUE

code/post_clear_check/src/post_clear_check/feishu.py

@@ -25,9 +25,11 @@ import httpx
 from loguru import logger
 
 from post_clear_check.chain import (
+    CARD_DIMENSION,
     CHAIN_BUY,
     CHAIN_SELL,
     CHAINS,
+    DIMENSION_NAME,
     OWNER_BATCH,
     OWNER_FUND,
     OWNER_MASTER,
@@ -232,8 +234,16 @@ def render_title(
     """一行标题：结论、链路名称、归属名称、主键值。不写系统和清算步骤。失败重发时末尾加「重发」。"""
 
     label = VERDICT_LABEL.get(verdict)
+    if label is None:
+        raise FeishuError("卡片结论或主键名称无法生成标题")
+    # 方向卡标题不写归属和营业日。营业日在正文第一行。
+    if chain_name in DIMENSION_NAME:
+        title = f"{DIMENSION_NAME[chain_name]} {label}"
+        if resend:
+            return f"{title} 重发"
+        return title
     key_name = KEY_NAME_BY_KIND.get(card_kind)
-    if label is None or key_name is None:
+    if key_name is None:
         raise FeishuError("卡片结论或主键名称无法生成标题")
     chain = CHAINS.get(chain_name)
     if chain is None:
@@ -865,6 +875,18 @@ def render_body(
     theme 是整张卡的总结论。错误主题只写错误清单。没传时，只有全部都是错误行才按错误主题写。
     """
 
+    if chain_name in DIMENSION_NAME:
+        from post_clear_check.dimension_body import render_dimension_body
+
+        return render_dimension_body(
+            chain_name=chain_name,
+            business_date=business_date,
+            trade_date=trade_date,
+            batch_no=batch_no,
+            lines=lines,
+            record_no=record_no,
+            extra=extra,
+        )
     chain = CHAINS.get(chain_name)
     if chain is None:
         raise FeishuError("卡片链路不在登记表之内")
@@ -1041,13 +1063,18 @@ def build_payload(
     return payload
 
 
-def body_lines(lines: Sequence[ConclusionView], verdict: str) -> tuple[ConclusionView, ...]:
+def body_lines(
+    lines: Sequence[ConclusionView],
+    verdict: str,
+    *,
+    keep_all: bool = False,
+) -> tuple[ConclusionView, ...]:
     """飞书正文实际写出的结论行。
 
-    主题为错误时只留错误行。同一张卡上的正确行和可疑行仍在库里，不写进这张卡的正文。
+    主题为错误时只留错误行。方向卡保留正确行，计算过程仍要能对上。
     """
 
-    if verdict == VERDICT_ERROR:
+    if verdict == VERDICT_ERROR and not keep_all:
         return tuple(line for line in lines if line.verdict == VERDICT_ERROR)
     return tuple(line for line in lines if line.verdict in VERDICT_LABEL)
 
@@ -1095,10 +1122,11 @@ def pack_bodies(
     signed: bool,
     limit: int = PAYLOAD_BYTE_LIMIT,
     sample_index: int | None = None,
+    keep_all: bool = False,
 ) -> tuple[str, ...]:
     """按真实请求体字节数装箱。错误主题只装错误行；放不下就拆成多张，单行仍超限再截短。"""
 
-    usable = body_lines(lines, verdict)
+    usable = body_lines(lines, verdict, keep_all=keep_all)
     if not usable:
         return ()
     common = {
@@ -1315,6 +1343,7 @@ async def send_card(
         lines=card.lines,
         signed=secret != "",
         sample_index=card.sample_index,
+        keep_all=card.card_kind == CARD_DIMENSION,
     )
     return await _deliver(
         settings,

code/post_clear_check/src/post_clear_check/orchestrate.py

@@ -48,21 +48,30 @@ from post_clear_check.gate import (
 )
 from post_clear_check.published import daycut_mismatches
 from post_clear_check.chain import (
+    CARD_DIMENSION,
     CHAIN_GATE,
-    CHAIN_ORDER,
     CHAIN_UNASSIGNED,
     CHAINS,
+    DIM_ISSUE,
+    DIM_ORDER,
+    DIM_SALES,
+    DIM_TRS,
+    DIMENSIONS,
+    DIMENSION_NAME,
     GATE_MISSING_TABLE,
+    INDEX_MARK,
     OWNER_BATCH,
     OWNER_FUND,
     OWNER_MASTER,
     OWNER_USER,
+    STATUS_MARK,
     SIDE_NONE,
     hang,
     hang_by_side,
     owner_of,
     unassign,
 )
+from post_clear_check.dimension_body import GATE_FORMULAS, dimension_of_gate
 from post_clear_check.issue_common import is_population_bucket
 from post_clear_check.issue_reports import check_issue_reports
 from post_clear_check.order_cell import CheckLine
@@ -100,11 +109,12 @@ CARD_FUND = OWNER_FUND
 CARD_USER = OWNER_USER
 CARD_BATCH = OWNER_BATCH
 
-_KIND_RANK = {CARD_MASTER: 0, CARD_FUND: 1, CARD_USER: 2, CARD_BATCH: 3}
+_KIND_RANK = {CARD_MASTER: 0, CARD_FUND: 1, CARD_USER: 2, CARD_BATCH: 3, CARD_DIMENSION: 4}
 # 结论行每次 executemany 的行数。只影响进度推进的粒度，单条语句长度由驱动按 1MB 切分。
 _LINE_BATCH = 2000
 _VERDICTS = frozenset({VERDICT_CORRECT, VERDICT_ERROR, VERDICT_SUSPECT})
-_KINDS = frozenset(set(_KIND_RANK))
+_KINDS = frozenset(set(_KIND_RANK) | {CARD_SUMMARY})
+_DIMENSION_INDEX = {code: index for index, code in enumerate(DIMENSIONS)}
 
 # 自有库语句。业务连接不执行这些写入。
 _FIND_RUN_SQL = (
@@ -236,49 +246,19 @@ def _worst_verdict(verdicts: Iterable[str]) -> str:
     return VERDICT_CORRECT
 
 
-def _summary_drafts(details: Sequence[tuple[str, str, str, str]]) -> tuple[CardDraft, ...]:
-    """每条有明细的链路一张概况。没有结论行。biz_key 用链路代码，便于幂等。
-
-    details 的四列是链路、归属、主键、明细卡总结论。
-    """
-
-    grouped: dict[str, list[str]] = {}
-    for chain_name, _kind, _biz_key, verdict in details:
-        grouped.setdefault(chain_name, []).append(verdict)
-    unknown = sorted(code for code in grouped if code not in CHAIN_ORDER)
-    if unknown:
-        logger.error("概况跳过未登记链路：{}", "、".join(unknown))
-    drafts: list[CardDraft] = []
-    for chain_name in sorted(grouped, key=lambda code: CHAIN_ORDER.get(code, len(CHAIN_ORDER))):
-        if chain_name not in CHAIN_ORDER:
-            continue
-        drafts.append(
-            CardDraft(
-                chain_name,
-                CARD_SUMMARY,
-                chain_name,
-                _worst_verdict(grouped[chain_name]),
-                (),
-            )
-        )
-    return tuple(drafts)
-
-
 def _attach_summaries(cards: Sequence[CardDraft]) -> tuple[CardDraft, ...]:
-    """概况排在明细前面，这样新轮次的主键也是概况在先。已带概况的再收一次不会重复。"""
+    """方向卡不再另发链路概况。旧轮次已经落下的概况卡保持原样，这里不补。"""
 
-    details = tuple(card for card in cards if card.card_kind != CARD_SUMMARY)
-    rows = tuple((card.chain_name, card.card_kind, card.biz_key, card.verdict) for card in details)
-    return _summary_drafts(rows) + details
+    return tuple(cards)
 
 
 def _pending_order(cards: Sequence[StoredCard]) -> tuple[StoredCard, ...]:
-    """先发概况，概况之间按链路登记顺序；再按主键升序发明细。旧轮次补插的概况主键更大，不能靠主键排序。"""
+    """四个方向按固定顺序先发。门禁快照和旧归属卡仍按读出的顺序跟在后面。"""
 
-    summaries = [card for card in cards if card.card_kind == CARD_SUMMARY]
-    details = [card for card in cards if card.card_kind != CARD_SUMMARY]
-    summaries.sort(key=lambda card: CHAIN_ORDER.get(card.chain_name, len(CHAIN_ORDER)))
-    return tuple(summaries) + tuple(details)
+    ranked = [card for card in cards if card.card_kind == CARD_DIMENSION and card.chain_name in _DIMENSION_INDEX]
+    rest = [card for card in cards if card not in ranked]
+    ranked.sort(key=lambda card: _DIMENSION_INDEX[card.chain_name])
+    return tuple(ranked) + tuple(rest)
 
 
 def dedupe_key(
@@ -327,61 +307,168 @@ def assemble_cards(
     missing_tables: Sequence[str] = (),
     gate_lines: Sequence[CheckLine] = (),
 ) -> tuple[CardDraft, ...]:
-    """按「链路 + 归属 + 主键」收卡。归属按链路登记的优先顺序取第一个非空键。
+    """按产生来源收成最多四张方向卡。内部链路代码留在结论行上。
 
-    没挂链、链路要求的归属键为空的行进未归链路，结论为错误。缺表进核对门禁。
+    没挂链或归属键为空的行仍改成未归错误，放进产生它的那个方向。
+    门禁行写进对应报表卡，不另发第五张。订单卡只在摘要里引用三端门禁。
     """
 
-    buckets: dict[tuple[str, str, str], list[CheckLine]] = {}
-    loose: list[CheckLine] = []
-
-    def add(chain_name: str, kind: str, biz_key: str, line: CheckLine) -> None:
-        buckets.setdefault((chain_name, kind, biz_key), []).append(line)
-
-    def place(line: CheckLine) -> None:
-        if line.chain_name == CHAIN_UNASSIGNED:
-            loose.append(line)
-            return
-        chain = CHAINS.get(line.chain_name)
-        if chain is None:
-            loose.append(unassign(line, "没有挂上链路"))
-            return
-        owner = owner_of(line)
-        if owner is None:
-            loose.append(unassign(line, f"{chain.name}要求的归属键为空"))
-            return
-        add(chain.code, owner[0], owner[1], line)
-
+    incoming = (
+        any(order_lines.values())
+        or bool(issue_lines)
+        or bool(sales_lines)
+        or bool(trs_lines)
+        or bool(missing_tables)
+        or bool(gate_lines)
+    )
+    if not incoming:
+        return ()
+    if business_date is None:
+        raise ValueError("方向卡缺少营业日")
+    day = business_date.isoformat()
+    buckets: dict[str, list[CheckLine]] = {code: [] for code in DIMENSIONS}
+    population: set[str] = set()
     for order_no in sorted(order_lines):
-        for line in order_lines[order_no]:
-            place(line)
-    for line in tuple(issue_lines) + tuple(sales_lines) + tuple(trs_lines):
-        place(line)
-    if loose or missing_tables:
-        if business_date is None:
-            raise ValueError("没有归属的结论行缺少营业日")
-        day = business_date.isoformat()
-        for line in loose:
-            add(CHAIN_UNASSIGNED, CARD_BATCH, day, line)
-        for table in missing_tables:
-            add(CHAIN_GATE, CARD_BATCH, day, _missing_table_line(table))
-    if gate_lines:
-        if business_date is None:
-            raise ValueError("核对门禁结论行缺少营业日")
-        day = business_date.isoformat()
-        for line in gate_lines:
-            add(CHAIN_GATE, CARD_BATCH, day, line)
+        folded = [_fold_source(line) for line in order_lines[order_no]]
+        if not folded:
+            continue
+        population.add(str(order_no))
+        buckets[DIM_ORDER].extend(folded)
+    for line in issue_lines:
+        buckets[DIM_ISSUE].append(_fold_source(line))
+    for line in sales_lines:
+        buckets[DIM_SALES].append(_fold_source(line))
+    for line in trs_lines:
+        buckets[DIM_TRS].append(_fold_source(line))
+    for table in missing_tables:
+        buckets[_table_dimension(table)].append(_missing_table_line(table))
+    for line in gate_lines:
+        buckets[dimension_of_gate(line.formula_text, line.check_text)].append(line)
+    if buckets[DIM_ORDER]:
+        buckets[DIM_ORDER] = [
+            *_status_lines(buckets),
+            *buckets[DIM_ORDER],
+            *_index_lines(population, buckets),
+        ]
     drafts: list[CardDraft] = []
-    # 同一归属对象的几条链路相邻，群里按主键能连着看。
-    order = sorted(buckets, key=lambda item: (_KIND_RANK[item[1]], item[2], CHAIN_ORDER[item[0]]))
-    for chain_name, kind, biz_key in order:
-        rows = tuple(buckets[(chain_name, kind, biz_key)])
+    for code in DIMENSIONS:
+        rows = tuple(buckets[code])
         if not rows:
             continue
-        drafts.append(CardDraft(chain_name, kind, biz_key, summarize(rows), rows))
+        drafts.append(CardDraft(code, CARD_DIMENSION, day, _dimension_verdict(rows), rows))
     return tuple(drafts)
 
 
+def _fold_source(line: CheckLine) -> CheckLine:
+    """归属键为空时改成未归错误。方向仍由调用方决定，不在这里改。"""
+
+    if line.chain_name == CHAIN_UNASSIGNED:
+        return line
+    chain = CHAINS.get(line.chain_name)
+    if chain is None:
+        return unassign(line, "没有挂上链路")
+    if chain.code == CHAIN_GATE:
+        return line
+    if owner_of(line) is None:
+        return unassign(line, f"{chain.name}要求的归属键为空")
+    return line
+
+
+def _table_dimension(table: str) -> str:
+    """1146 按库名前缀归到报表卡。认不出的表归发行报表，避免丢掉。"""
+
+    if table.startswith("db_trs.") or table.startswith("trs."):
+        return DIM_TRS
+    if table.startswith("db_sales.") or table.startswith("sales."):
+        return DIM_SALES
+    return DIM_ISSUE
+
+
+def _annotation(line: CheckLine) -> bool:
+    return line.formula_text.startswith(INDEX_MARK) or line.formula_text.startswith(STATUS_MARK)
+
+
+def _dimension_verdict(lines: Sequence[CheckLine]) -> str:
+    """索引和批次状态不抬高、也不压低这张卡的总结论。"""
+
+    scored = [line for line in lines if not _annotation(line)]
+    if not scored:
+        return VERDICT_CORRECT
+    return summarize(scored)
+
+
+def _status_lines(buckets: Mapping[str, Sequence[CheckLine]]) -> list[CheckLine]:
+    """订单摘要里的三端门禁一行。正文不重复门禁公式。"""
+
+    found: list[CheckLine] = []
+    for code, label in ((DIM_ISSUE, "发行"), (DIM_SALES, "销售"), (DIM_TRS, "TRS")):
+        count = sum(1 for line in buckets[code] if line.formula_text in GATE_FORMULAS or line.chain_name == CHAIN_GATE)
+        if count:
+            text = f"{label}门禁：错误 {count} 条。明细在{DIMENSION_NAME[code]}。"
+        else:
+            text = f"{label}门禁：无差异。明细在{DIMENSION_NAME[code]}。"
+        found.append(
+            CheckLine(
+                formula_text=f"{STATUS_MARK}{label}",
+                calc_text=text,
+                calc_value=None,
+                check_text=text,
+                check_value=None,
+                basis="",
+                verdict=VERDICT_CORRECT,
+                chain_name=CHAIN_GATE,
+            )
+        )
+    return found
+
+
+def _index_lines(population: set[str], buckets: Mapping[str, Sequence[CheckLine]]) -> list[CheckLine]:
+    """本轮主单在三张报表卡上的对错笔数。没有主单号的账户级报表不收进来。"""
+
+    found: list[CheckLine] = []
+    for code, label in ((DIM_ISSUE, "发行"), (DIM_SALES, "销售"), (DIM_TRS, "TRS")):
+        grouped: dict[str, dict[str, str]] = {}
+        for line in buckets[code]:
+            if not line.report_code or line.master_order_no not in population:
+                continue
+            orders = grouped.setdefault(line.report_code, {})
+            orders[line.master_order_no] = _worse_verdict(orders.get(line.master_order_no), line.verdict)
+        for report_code in sorted(grouped):
+            orders = grouped[report_code]
+            counts = {VERDICT_CORRECT: 0, VERDICT_ERROR: 0, VERDICT_SUSPECT: 0}
+            for verdict in orders.values():
+                counts[verdict] += 1
+            bits = [f"{label} {report_code}：本轮主单 {len(orders)} 笔"]
+            for verdict, name in (
+                (VERDICT_CORRECT, "正确"),
+                (VERDICT_ERROR, "错误"),
+                (VERDICT_SUSPECT, "可疑"),
+            ):
+                if counts[verdict]:
+                    bits.append(f"{name} {counts[verdict]}")
+            text = "，".join(bits) + f"。明细在{DIMENSION_NAME[code]}。"
+            found.append(
+                CheckLine(
+                    formula_text=f"{INDEX_MARK}{report_code}",
+                    calc_text=text,
+                    calc_value=None,
+                    check_text=text,
+                    check_value=None,
+                    basis="",
+                    verdict=VERDICT_CORRECT,
+                    report_code=report_code,
+                )
+            )
+    return found
+
+
+def _worse_verdict(current: str | None, verdict: str) -> str:
+    rank = {VERDICT_ERROR: 0, VERDICT_SUSPECT: 1, VERDICT_CORRECT: 2}
+    if current is None or rank[verdict] < rank[current]:
+        return verdict
+    return current
+
+
 def assemble_batch_cards(lines: Sequence[ConclusionLine]) -> tuple[CardDraft, ...]:
     """批次门禁一条一个业务键。同一个批次号上的多条放在同一张卡，归核对门禁。"""
 
@@ -550,38 +637,6 @@ class SqlRunStore:
         rows = await self._query(_SELECT_RUN_RESULTS_SQL, (run_id,))
         return tuple((str(row[0]), str(row[1]), str(row[2]), str(row[3])) for row in rows)
 
-    async def ensure_summaries(self, run_id: int, ready: ReadyRun, now: datetime) -> None:
-        """旧轮次没有概况卡时补插。不改三个计数，不复制明细的结论行。
-
-        和明细写在不同的事务里：补插失败不会把已经落下的明细卷走。
-        幂等键撞上说明另一个进程刚补过，回滚后继续发。
-        """
-
-        rows = await self.detail_rows(run_id)
-        have = {chain for kind, chain, _biz, _verdict in rows if kind == CARD_SUMMARY}
-        details = tuple(
-            (chain, kind, biz, verdict)
-            for kind, chain, biz, verdict in rows
-            if kind != CARD_SUMMARY
-        )
-        fresh = [card for card in _summary_drafts(details) if card.chain_name not in have]
-        if not fresh:
-            return
-        await self._connection.begin()
-        try:
-            for card in fresh:
-                await self._insert_card(run_id, ready, card, now, None)
-            await self._connection.commit()
-        except IntegrityError as exc:
-            await self._rollback_quietly()
-            if _duplicate(exc):
-                return
-            raise
-        except Exception:
-            await self._rollback_quietly()
-            raise
-        logger.info("补插链路概况 {} 张", len(fresh))
-
     async def set_notify(self, result_id: int, status: str, now: datetime) -> None:
         """把一张卡改成 sent 或 failed。只按主键更新。"""
 
@@ -851,9 +906,8 @@ async def _process_ready(
         normalized = _attach_summaries(
             tuple(item for item in (_normalize_card(card) for card in cards) if item is not None)
         )
-        detail_count = sum(1 for card in normalized if card.card_kind != CARD_SUMMARY)
         stage.summary = (
-            f"明细 {detail_count} 张，概况 {len(normalized) - detail_count} 张，"
+            f"方向 {len(normalized)} 张，"
             f"结论行 {sum(len(card.lines) for card in normalized)} 条"
         )
     await _process_existing_or_save(settings, store, ready, normalized, send_fn, business=business)
@@ -887,8 +941,7 @@ async def _process_existing_or_save(
     total_lines = sum(len(card.lines) for card in cards)
     async with Stage("写入自有库", total=total_lines) as stage:
         run_id = await store.save_new_run(ready, cards, _now(), tick=stage.advance)
-        detail_count = sum(1 for card in cards if card.card_kind != CARD_SUMMARY)
-        stage.summary = f"明细 {detail_count} 张，概况 {len(cards) - detail_count} 张，结论行 {total_lines} 条"
+        stage.summary = f"方向 {len(cards)} 张，结论行 {total_lines} 条"
     if run_id is None:
         found = await store.find_run_id(ready)
         if found is None:
@@ -914,8 +967,6 @@ async def _send_unsent(
     """先补齐并发送概况，再按主键升序发送明细。成功改 sent，失败改 failed，不新插结论行。"""
 
     async with Stage("发送飞书卡片") as stage:
-        stage.note("补齐链路概况")
-        await store.ensure_summaries(run_id, ready, _now())
         stage.note("读取待发送的卡片")
         cards = _pending_order(await store.unsent_cards(run_id))
         rows = await store.detail_rows(run_id)
@@ -992,12 +1043,14 @@ async def _send_one(
     """
 
     summary = card.card_kind == CARD_SUMMARY
+    # 四个方向各只有一张。按结论类型或条数截断会直接少掉一个方向。
+    dimension = card.card_kind == CARD_DIMENSION
     if summary:
         if not owners:
             return "skipped"
     elif not card.lines:
         return "skipped"
-    elif allowed is not None and card.verdict not in allowed:
+    elif not dimension and allowed is not None and card.verdict not in allowed:
         label = _VERDICT_LABEL.get(card.verdict, card.verdict)
         logger.info(
             "飞书卡片不推送，{} {} {}，结论 {} 不在推送配置内",
@@ -1007,7 +1060,7 @@ async def _send_one(
             label,
         )
         return "skipped"
-    elif not summary and limit is not None:
+    elif not summary and not dimension and limit is not None:
         taken = 0 if used is None else used.get(card.chain_name, 0)
         if taken >= limit:
             logger.info(
@@ -1022,7 +1075,7 @@ async def _send_one(
             used[card.chain_name] = taken + 1
     shown = batch_no if batch_no is not None else card.biz_key
     sample_index = None
-    if not summary and limit is not None and used is not None:
+    if not summary and not dimension and limit is not None and used is not None:
         sample_index = used[card.chain_name]
     outbound = OutboundCard(
         card_kind=card.card_kind,
@@ -1185,12 +1238,16 @@ def _synthetic_run(market: str, trade_date: date, business_date: date) -> ReadyR
 
 def _normalize_card(card: CardDraft) -> CardDraft | None:
     if card.card_kind not in _KINDS:
-        raise ValueError("卡片种类不在四种之内")
-    chain = CHAINS.get(card.chain_name)
-    if chain is None:
-        raise ValueError("卡片链路不在登记表之内")
-    if card.card_kind not in chain.owners:
-        raise ValueError("卡片归属不在该链路登记的归属之内")
+        raise ValueError("卡片种类不在登记的种类之内")
+    if card.card_kind == CARD_DIMENSION:
+        if card.chain_name not in _DIMENSION_INDEX:
+            raise ValueError("方向代码不在四个方向之内")
+    else:
+        chain = CHAINS.get(card.chain_name)
+        if chain is None:
+            raise ValueError("卡片链路不在登记表之内")
+        if card.card_kind != CARD_SUMMARY and card.card_kind not in chain.owners:
+            raise ValueError("卡片归属不在该链路登记的归属之内")
     stored, original = _fit_biz_key(card.biz_key)
     raw_lines = card.lines
     if original and raw_lines:
@@ -1241,7 +1298,7 @@ def _clip(text: str, limit: int) -> str:
 
 
 def _counts(cards: Sequence[CardDraft]) -> tuple[int, int, int]:
-    """按明细卡片数计，不按结论行数，也不把链路概况算进去。"""
+    """按方向卡计数。门禁快照仍单独计。链路概况不进这三个数。"""
 
     correct = error = suspect = 0
     for card in cards:

code/post_clear_check/tests/test_chain.py

@@ -7,6 +7,8 @@ from decimal import Decimal
 import pytest
 
 from post_clear_check.chain import (
+    CHAIN_DIMENSIONS,
+    DIMENSIONS,
     BASIS_DESIGN,
     BASIS_EXCEL,
     BASIS_PROTOTYPE,
@@ -422,3 +424,13 @@ def test_cash_change_keeps_master_order_but_belongs_to_fund() -> None:
     assert lines
     assert all(line.master_order_no == "ORD9" for line in lines)
     assert all(owner_of(line) == ("fund_account", "FA1") for line in lines)
+
+
+def test_every_internal_chain_maps_to_a_dimension() -> None:
+    """11 条内部链路和未归链路都能归进四个方向。方向表不替代挂链。"""
+
+    assert set(CHAIN_DIMENSIONS) == set(CHAINS)
+    for code, targets in CHAIN_DIMENSIONS.items():
+        assert targets
+        assert targets <= set(DIMENSIONS)
+        assert code in CHAINS

code/post_clear_check/tests/test_feishu.py

@@ -15,6 +15,7 @@ from loguru import logger
 from pydantic import SecretStr
 
 from post_clear_check.chain import (
+    CARD_DIMENSION,
     CHAIN_BUY,
     CHAIN_CASH_CHANGE,
     CHAIN_GATE,
@@ -22,6 +23,8 @@ from post_clear_check.chain import (
     CHAIN_TRADE,
     CHAIN_UNASSIGNED,
     CHAINS,
+    DIM_ORDER,
+    DIM_SALES,
 )
 from post_clear_check.config import Settings
 from post_clear_check.feishu import (
@@ -1259,3 +1262,160 @@ def test_summary_without_owners_is_not_posted() -> None:
     result = asyncio.run(send_summary(_settings(), card, transport=_transport(handler), gate=_gate()))
     assert result.attempted is False
     assert result.ok is False
+
+
+@dataclass(frozen=True)
+class _DimLine:
+    """方向卡正文要读到的归属和报表编号。"""
+
+    formula_text: str
+    calc_text: str
+    calc_value: Decimal | None
+    check_text: str
+    check_value: Decimal | None
+    basis: str
+    verdict: str
+    report_code: str = ""
+    master_order_no: str = ""
+    fund_account: str = ""
+    user_account: str = ""
+
+
+def _dim(
+    formula: str,
+    verdict: str,
+    *,
+    calc: str = "1 = 1",
+    check: str = "列 = 1",
+    calc_value: Decimal | None = Decimal("1"),
+    check_value: Decimal | None = Decimal("1"),
+    basis: str = "依据甲",
+    report_code: str = "",
+    master_order_no: str = "",
+    fund_account: str = "",
+) -> _DimLine:
+    return _DimLine(
+        formula_text=formula,
+        calc_text=calc,
+        calc_value=calc_value,
+        check_text=check,
+        check_value=check_value,
+        basis=basis,
+        verdict=verdict,
+        report_code=report_code,
+        master_order_no=master_order_no,
+        fund_account=fund_account,
+    )
+
+
+def test_dimension_title_is_name_and_verdict() -> None:
+    """方向卡标题只写方向和结论。重发加在末尾。"""
+
+    assert render_title(VERDICT_SUSPECT, DIM_ORDER, CARD_DIMENSION, "2026-10-30") == "订单链路 可疑"
+    assert render_title(VERDICT_CORRECT, DIM_SALES, CARD_DIMENSION, "2026-10-30") == "销售报表 正确"
+    assert render_title(
+        VERDICT_ERROR, DIM_ORDER, CARD_DIMENSION, "2026-10-30", resend=True
+    ) == "订单链路 错误 重发"
+
+
+def test_dimension_order_body_collapses_correct_fields() -> None:
+    """同一字段的正确行收成一块，并写上笔数。错误行单独展开，不并进这块。"""
+
+    body = render_body(
+        business_date=DAY,
+        trade_date=DAY,
+        batch_no="IB",
+        chain_name=DIM_ORDER,
+        record_no="8",
+        lines=(
+            _dim("票据面值", VERDICT_CORRECT, master_order_no="ORD1", calc="100 × 10% = 10"),
+            _dim("票据面值", VERDICT_CORRECT, master_order_no="ORD2", calc="100 × 10% = 10"),
+            _dim("订单卖出股数", VERDICT_SUSPECT, master_order_no="ORD9", calc_value=Decimal("3"), check_value=Decimal("9")),
+            _dim("关联报表索引：OTC003", VERDICT_CORRECT, calc="发行 OTC003：本轮主单 2 笔，正确 2。明细在发行报表。", check=""),
+            _dim("批次状态：发行", VERDICT_CORRECT, calc="发行门禁：无差异。明细在发行报表。", check="", calc_value=None, check_value=None),
+        ),
+    )
+    assert body.startswith("营业日：2026-10-15\n交易日：2026-10-15\n批次号：IB")
+    assert "涉及：" not in body
+    assert "人口：已完成或已结算的 ORD 主单。本方向正确 2 笔，错误 0 笔，可疑 1 笔。" in body
+    assert "发行门禁：无差异。明细在发行报表。" in body
+    assert "每单链路\n本轮没有执行事实或对冲执行根结论。" in body
+    assert "盘中公式\n字段：票据面值" in body
+    assert "清算对照\n本轮没有这类结论。" in body
+    assert "关联报表\n字段：发行 OTC003" in body
+    assert "结论：正确（2/2）" in body
+    assert "实据：ORD1、ORD2 无差异" in body
+    assert "实据：本轮主单 2 笔，正确 2" in body
+    assert "节点：订单卖出股数" in body
+    assert "结论：可疑。" in body
+    assert "错误清单：无" in body
+    assert "正确另有" not in body
+
+
+def test_dimension_report_body_keeps_correct_rows_on_error_theme() -> None:
+    """报表卡即使总结论是错误，正确列仍留在分报表，错误行逐条写期望和实据。"""
+
+    body = render_body(
+        business_date=DAY,
+        trade_date=DAY,
+        batch_no="SB",
+        chain_name=DIM_SALES,
+        record_no="9",
+        theme=VERDICT_ERROR,
+        lines=(
+            _dim(
+                "现金余额",
+                VERDICT_CORRECT,
+                check="OTC001 资金账户资金报表.现金余额 = 1",
+                report_code="OTC001",
+                fund_account="FA1",
+            ),
+            _dim(
+                "买入待交收流水金额 = −订单总金额",
+                VERDICT_ERROR,
+                calc="−220421.76",
+                check="OTC002 现金变动记录.金额 = -220400.00（资金账户 FA00000001）",
+                calc_value=Decimal("-220421.76"),
+                check_value=Decimal("-220400.00"),
+                report_code="OTC002",
+                fund_account="FA00000001",
+                master_order_no="ORD001",
+                basis="买入订单第 28 行",
+            ),
+        ),
+    )
+    assert "报表：OTC001 资金账户资金报表" in body
+    assert "筛选：营业日 2026-10-15，资金账户 FA1" in body
+    assert "期望：现金余额" in body
+    assert "实据：1 列无差异" in body
+    assert "结论：正确" in body
+    assert "正确与可疑留在核对明细" not in body
+    assert "节点：买入待交收流水金额 = −订单总金额" in body
+    assert "期望：−220421.76" in body
+    assert "当前 -220400.00" in body or "当前 -220400" in body
+    assert "不符合预期" in body
+    assert "筛选：主订单 ORD001 · 资金账户 FA00000001" in body
+
+
+def test_dimension_long_body_keeps_fields_and_errors() -> None:
+    """方向卡不按 3500 字删掉盘中公式。错误行和字段都留在正文里。"""
+
+    correct = tuple(
+        _dim(f"字段{index}", VERDICT_CORRECT, basis="依" * 80, master_order_no=f"ORD{index}")
+        for index in range(40)
+    )
+    error = _dim("金额不符", VERDICT_ERROR, calc_value=Decimal("1"), check_value=Decimal("2"), master_order_no="ORDX")
+    body = render_body(
+        business_date=DAY,
+        trade_date=DAY,
+        batch_no="IB",
+        chain_name=DIM_ORDER,
+        record_no="11",
+        lines=(*correct, error),
+    )
+    assert "字段：字段0" in body
+    assert "字段：字段39" in body
+    assert "节点：金额不符" in body
+    assert "不符合预期" in body
+    assert "正确另有" not in body
+    assert len(body) > BODY_LIMIT

code/post_clear_check/tests/test_orchestrate.py

@@ -9,6 +9,7 @@ from decimal import Decimal
 from asyncmy.errors import IntegrityError
 
 from post_clear_check.chain import (
+    CARD_DIMENSION,
     CHAIN_ACCOUNT_ASSET,
     CHAIN_BUY,
     CHAIN_CASH_CHANGE,
@@ -18,6 +19,10 @@ from post_clear_check.chain import (
     CHAIN_TRADE,
     CHAIN_UNASSIGNED,
     CHAIN_USER_ASSET,
+    DIM_ISSUE,
+    DIM_ORDER,
+    DIM_SALES,
+    DIM_TRS,
     UNASSIGNED_BASIS,
 )
 from post_clear_check.feishu import CARD_SUMMARY, SendResult
@@ -380,8 +385,8 @@ def test_same_round_second_entry_does_not_insert_another_run() -> None:
     assert {row["status"] for row in conn.results} == {"sent"}
 
 
-def test_cards_split_by_chain_owner_and_key() -> None:
-    """按「链路 + 归属 + 主键」分卡。同一主订单的两条链路是两张卡，资金变动行归资金账户。"""
+def test_cards_group_into_four_dimensions() -> None:
+    """一轮最多四张方向卡。报表行留在产生它的那一端，订单卡只做索引。"""
 
     conn = _Own()
     sent: list[object] = []
@@ -390,53 +395,29 @@ def test_cards_split_by_chain_owner_and_key() -> None:
         await _run(conn, _script(), [], sent)
 
     asyncio.run(scenario())
-    assert _formulas(conn, "ORD-A", CHAIN_BUY) == ["订单A", "发行A"]
-    assert _formulas(conn, "ORD-A", CHAIN_TRADE) == ["TRS002-A"]
-    assert len(_cards_of(conn, "ORD-A")) == 2
-    assert _formulas(conn, "ORD-A1", CHAIN_BUY) == ["别的编号"]
-    assert _formulas(conn, "ORD-B", CHAIN_SELL) == ["订单B", "发行B"]
-    assert _formulas(conn, "ORD-B", CHAIN_TRADE) == ["TRS-HR99"]
-    # 资金变动归资金账户：行上同时有主订单编号也不进主订单卡。
-    assert "销售带订单A" not in _formulas(conn, "ORD-A")
-    assert _formulas(conn, "FA9", CHAIN_CASH_CHANGE) == ["销售带订单A"]
-    assert _formulas(conn, "FA9", CHAIN_ACCOUNT_ASSET) == ["资金A"]
-    assert _formulas(conn, "FA9", CHAIN_NOTE_HOLDING) == ["切片1", "切片2"]
-    assert _formulas(conn, "U9", CHAIN_USER_ASSET) == ["用户A"]
-    # 委托执行没有主订单编号、也没有用户账号，进未归链路，结论为错误。
-    unassigned = _cards_of(conn, "2026-10-17", CHAIN_UNASSIGNED)
-    assert [row["kind"] for row in unassigned] == ["clearing_batch"]
-    assert unassigned[0]["verdict"] == "error"
-    assert _formulas(conn, "2026-10-17", CHAIN_UNASSIGNED) == ["结论行没有归属：HR150不算"]
-    detail_chains = [
-        CHAIN_BUY,
-        CHAIN_TRADE,
-        CHAIN_BUY,
-        CHAIN_SELL,
-        CHAIN_TRADE,
-        CHAIN_CASH_CHANGE,
-        CHAIN_ACCOUNT_ASSET,
-        CHAIN_NOTE_HOLDING,
-        CHAIN_USER_ASSET,
-        CHAIN_UNASSIGNED,
-    ]
-    # 概况按链路登记顺序先发，同一条链路只一张。明细顺序不变，排在概况后面。
-    summary_chains = [
-        CHAIN_BUY,
-        CHAIN_SELL,
-        CHAIN_CASH_CHANGE,
-        CHAIN_ACCOUNT_ASSET,
-        CHAIN_NOTE_HOLDING,
-        CHAIN_TRADE,
-        CHAIN_USER_ASSET,
-        CHAIN_UNASSIGNED,
-    ]
-    assert [card.chain_name for card in sent] == summary_chains + detail_chains
-    assert [card.card_kind for card in sent[: len(summary_chains)]] == [CARD_SUMMARY] * len(summary_chains)
-    assert [card.record_no for card in sent] == [str(index) for index in range(1, len(sent) + 1)]
+    day = DAY.isoformat()
+    assert [card.chain_name for card in sent] == [DIM_ORDER, DIM_ISSUE, DIM_SALES, DIM_TRS]
+    assert [card.card_kind for card in sent] == [CARD_DIMENSION] * 4
+    assert [card.biz_key for card in sent] == [day] * 4
+    assert [card.record_no for card in sent] == ["1", "2", "3", "4"]
     assert sent[0].batch_no == "IB-9"
-    stored = _cards_of(conn, "ORD-A", CHAIN_BUY)[0]
-    assert stored["dedupe"] == dedupe_key(READY, "master_order", "ORD-A", chain_name=CHAIN_BUY)
+    order_formulas = _formulas(conn, day, DIM_ORDER)
+    assert "订单A" in order_formulas
+    assert "订单B" in order_formulas
+    assert "发行A" not in order_formulas
+    assert "销售带订单A" not in order_formulas
+    assert any(text.startswith("关联报表索引：OTC003") for text in order_formulas)
+    assert any("发行 OTC003：本轮主单 2 笔，正确 2" in row["calc_text"] for row in conn.lines)
+    assert "别的编号" not in order_formulas
+    assert _formulas(conn, day, DIM_ISSUE) == ["发行A", "发行B", "别的编号"]
+    assert _formulas(conn, day, DIM_SALES) == ["资金A", "切片1", "切片2", "销售带订单A"]
+    assert "结论行没有归属：HR150不算" in _formulas(conn, day, DIM_TRS)
+    assert "用户A" in _formulas(conn, day, DIM_TRS)
+    stored = _cards_of(conn, day, DIM_ORDER)[0]
+    assert stored["dedupe"] == dedupe_key(READY, CARD_DIMENSION, day, chain_name=DIM_ORDER)
     assert str(stored["dedupe"]).split("\x1f")[-1] == rule_pack_version()
+    assert conn.runs[0]["correct"] == 3
+    assert conn.runs[0]["error"] == 1
 
 
 def test_card_without_lines_is_not_inserted() -> None:
@@ -474,9 +455,7 @@ def test_failed_send_rerun_does_not_copy_lines() -> None:
     async def sender(_settings: object, card: object) -> SendResult:
         calls.append(card.record_no)
         resent.append((card.card_kind, card.resend))
-        # 概况一次成功。明细第一次失败，重发时才成功，用来钉住「不复制结论行」。
-        if card.card_kind == CARD_SUMMARY:
-            return SendResult(ok=True, retryable=False, attempted=True)
+        # 方向卡第一次失败，重发才成功，用来钉住「不复制结论行」。
         ok = card.resend
         return SendResult(ok=ok, retryable=not ok, attempted=True)
 
@@ -487,22 +466,19 @@ def test_failed_send_rerun_does_not_copy_lines() -> None:
         script["sales"] = lambda _data: []
         script["trs"] = lambda _data: []
         await _run(conn, script, sender=sender)
-        assert len(conn.lines) == 1
-        detail = next(row for row in conn.results if row["kind"] == "master_order")
+        assert sum(1 for row in conn.lines if row["formula"] == "只有一行") == 1
+        detail = next(row for row in conn.results if row["kind"] == CARD_DIMENSION)
         assert detail["status"] == "failed"
         await _run(conn, script, sender=sender)
         await _run(conn, script, sender=sender)
 
     asyncio.run(scenario())
     assert len(conn.runs) == 1
-    assert len(conn.lines) == 1
-    assert conn.lines[0]["formula"] == "只有一行"
-    detail = next(row for row in conn.results if row["kind"] == "master_order")
-    summary = next(row for row in conn.results if row["kind"] == CARD_SUMMARY)
-    assert calls == [str(summary["id"]), str(detail["id"]), str(detail["id"])]
-    assert resent == [(CARD_SUMMARY, False), ("master_order", False), ("master_order", True)]
+    assert sum(1 for row in conn.lines if row["formula"] == "只有一行") == 1
+    detail = next(row for row in conn.results if row["kind"] == CARD_DIMENSION)
+    assert calls == [str(detail["id"]), str(detail["id"])]
+    assert resent == [(CARD_DIMENSION, False), (CARD_DIMENSION, True)]
     assert detail["status"] == "sent"
-    assert summary["status"] == "sent"
 
 
 def test_verdict_is_error_when_suspect_and_error_share_a_card() -> None:
@@ -524,10 +500,10 @@ def test_verdict_is_error_when_suspect_and_error_share_a_card() -> None:
         await _run(conn, script)
 
     asyncio.run(scenario())
-    details = [row for row in conn.results if row["kind"] == "master_order"]
+    details = [row for row in conn.results if row["kind"] == CARD_DIMENSION]
     assert len(details) == 1
     assert details[0]["verdict"] == "error"
-    assert len([row for row in conn.results if row["kind"] == CARD_SUMMARY]) == 1
+    assert details[0]["chain"] == DIM_ORDER
     assert conn.runs[0]["error"] == 1
     assert conn.runs[0]["suspect"] == 0
     assert conn.runs[0]["correct"] == 0
@@ -691,11 +667,12 @@ def test_assemble_puts_same_fund_notes_on_one_card() -> None:
             ),
         ),
         (),
+        business_date=DAY,
     )
     assert len(cards) == 1
-    assert cards[0].chain_name == CHAIN_NOTE_HOLDING
-    assert cards[0].card_kind == "fund_account"
-    assert cards[0].biz_key == "FA9"
+    assert cards[0].chain_name == DIM_SALES
+    assert cards[0].card_kind == CARD_DIMENSION
+    assert cards[0].biz_key == DAY.isoformat()
     assert cards[0].verdict == "suspect"
     assert [line.formula_text for line in cards[0].lines] == ["切片1", "切片2"]
 
@@ -859,7 +836,7 @@ async def _run(
 
 
 def test_gate_lines_share_the_batch_card() -> None:
-    """已完成轮次的是/否检查并进这一天的核对门禁，不另开一张 completed 卡。"""
+    """已完成轮次的是/否检查并进对应报表卡，不另开一张方向卡，也不再发核对门禁卡。"""
 
     cards = assemble_cards(
         {},
@@ -870,8 +847,8 @@ def test_gate_lines_share_the_batch_card() -> None:
         gate_lines=(_line(FORMULA_ISSUE_PUBLISH, "应晚于本批营业日", verdict="error"),),
     )
     assert len(cards) == 1
-    assert cards[0].chain_name == CHAIN_GATE
-    assert cards[0].card_kind == "clearing_batch"
+    assert cards[0].chain_name == DIM_ISSUE
+    assert cards[0].card_kind == CARD_DIMENSION
     assert cards[0].biz_key == DAY.isoformat()
     assert cards[0].lines[0].formula_text == FORMULA_ISSUE_PUBLISH
     assert "completed" not in cards[0].lines[0].formula_text
@@ -888,8 +865,8 @@ def test_line_without_chain_is_kept_as_error() -> None:
         business_date=DAY,
     )
     assert len(cards) == 1
-    assert cards[0].chain_name == CHAIN_UNASSIGNED
-    assert cards[0].card_kind == "clearing_batch"
+    assert cards[0].chain_name == DIM_ISSUE
+    assert cards[0].card_kind == CARD_DIMENSION
     assert cards[0].biz_key == DAY.isoformat()
     assert cards[0].verdict == "error"
     line = cards[0].lines[0]
@@ -917,7 +894,7 @@ def test_line_with_empty_owner_key_is_kept_as_error() -> None:
         business_date=DAY,
     )
     assert [(card.chain_name, card.card_kind, card.biz_key) for card in cards] == [
-        (CHAIN_UNASSIGNED, "clearing_batch", DAY.isoformat())
+        (DIM_ISSUE, CARD_DIMENSION, DAY.isoformat())
     ]
     assert cards[0].verdict == "error"
     assert "票据存续要求的归属键为空" in cards[0].lines[0].calc_text
@@ -939,8 +916,12 @@ def test_trade_execution_falls_back_to_user_account() -> None:
                 chain_name=CHAIN_TRADE,
             ),
         ),
+        business_date=DAY,
     )
-    assert [(card.chain_name, card.card_kind, card.biz_key) for card in cards] == [(CHAIN_TRADE, "trs_user", "U9")]
+    assert [(card.chain_name, card.card_kind, card.biz_key) for card in cards] == [
+        (DIM_TRS, CARD_DIMENSION, DAY.isoformat())
+    ]
+    assert cards[0].lines[0].user_account == "U9"
 
 
 def test_order_rules_follow_master_direction() -> None:
@@ -952,8 +933,11 @@ def test_order_rules_follow_master_direction() -> None:
     assert lines
     assert {line.chain_name for line in lines} == {CHAIN_SELL}
     assert {line.master_order_no for line in lines} == {"ORD-S"}
-    cards = assemble_cards(collect_order_lines(result), (), (), ())
-    assert [(card.chain_name, card.card_kind, card.biz_key) for card in cards] == [(CHAIN_SELL, "master_order", "ORD-S")]
+    cards = assemble_cards(collect_order_lines(result), (), (), (), business_date=DAY)
+    assert [(card.chain_name, card.card_kind, card.biz_key) for card in cards] == [
+        (DIM_ORDER, CARD_DIMENSION, DAY.isoformat())
+    ]
+    assert {line.chain_name for line in cards[0].lines if line.chain_name} >= {CHAIN_SELL}
 
 
 def test_reference_orders_do_not_run_order_rules() -> None:
@@ -1059,10 +1043,9 @@ def test_missing_table_becomes_error_card() -> None:
     asyncio.run(scenario())
     assert [row["formula"] for row in conn.lines] == ["业务表应存在"]
     assert conn.lines[0]["check_text"] == "db_issue.report_trade_order"
-    detail = next(row for row in conn.results if row["kind"] == "clearing_batch")
+    detail = next(row for row in conn.results if row["kind"] == CARD_DIMENSION)
     assert detail["verdict"] == "error"
-    assert detail["chain"] == CHAIN_GATE
-    assert detail["kind"] == "clearing_batch"
+    assert detail["chain"] == DIM_ISSUE
     assert conn.lines[0]["basis"].startswith("研发设计 ")
 
 
@@ -1102,42 +1085,46 @@ def test_historical_trs_block_is_stored_once() -> None:
     assert len(conn.runs) == 1
     details = [row for row in conn.results if row["kind"] == "clearing_batch"]
     assert len(details) == 1
-    assert len([row for row in conn.results if row["kind"] == CARD_SUMMARY]) == 1
+    assert [row for row in conn.results if row["kind"] == CARD_SUMMARY] == []
     assert details[0]["verdict"] == "error"
     assert details[0]["biz"] == DAY.isoformat()
     assert details[0]["chain"] == CHAIN_GATE
 
 
-def test_push_filter_leaves_other_verdicts_pending() -> None:
-    """只配置错误时，正确卡不发，仍保持 pending，改配置后可以补发。"""
+def test_push_filter_does_not_drop_a_dimension() -> None:
+    """只配置错误时，四个方向仍然整张发送。截断一个方向会让正确的报表看不见。"""
 
     conn = _Own()
     sent: list[object] = []
-    settings = sample_settings().model_copy(update={"feishu_push": "错误"})
+    settings = sample_settings().model_copy(update={"feishu_push": "错误", "feishu_push_limit": 1})
 
     async def scenario() -> None:
         script = _script()
-        script["orders"] = lambda _result: {
-            "ORD-A": [_order_line("对", "correct", "ORD-A")],
-            "ORD-B": [_order_line("错", "error", "ORD-B")],
-        }
+        script["orders"] = lambda _result: {"ORD-A": [_order_line("对", "correct", "ORD-A")]}
         script["issue"] = lambda _data: []
-        script["sales"] = lambda _data: []
+        script["sales"] = lambda _data: [
+            _line(
+                "资金对",
+                "OTC001 资金账户资金报表.余额 = 1.00（资金账户 FA9）",
+                "correct",
+                report_code="OTC001",
+                fund_account="FA9",
+                chain_name=CHAIN_ACCOUNT_ASSET,
+            )
+        ]
         script["trs"] = lambda _data: []
         await _run(conn, script, sent=sent, settings=settings)
 
     asyncio.run(scenario())
-    # 概况先发，而且不看推送配置。明细里只有错误卡发出，正确卡仍是 pending。
-    assert [card.biz_key for card in sent] == [CHAIN_BUY, "ORD-B"]
-    assert [card.card_kind for card in sent] == [CARD_SUMMARY, "master_order"]
-    status = {row["biz"]: row["status"] for row in conn.results}
-    assert status["ORD-A"] == "pending"
-    assert status["ORD-B"] == "sent"
-    assert status[CHAIN_BUY] == "sent"
+    assert [card.chain_name for card in sent] == [DIM_ORDER, DIM_SALES]
+    assert [card.card_kind for card in sent] == [CARD_DIMENSION, CARD_DIMENSION]
+    assert all(card.sample_index is None for card in sent)
+    status = {(row["chain"], row["status"]) for row in conn.results}
+    assert status == {(DIM_ORDER, "sent"), (DIM_SALES, "sent")}
 
 
-def test_push_limit_samples_each_chain() -> None:
-    """每条链路只发待发列表最前面的张数。另一条链路各自再抽，不共用名额。"""
+def test_push_limit_does_not_split_a_dimension() -> None:
+    """条数上限不再把一个方向拆成只发第一张。同一张订单卡里的四笔一起发出。"""
 
     conn = _Own()
     sent: list[object] = []
@@ -1160,35 +1147,11 @@ def test_push_limit_samples_each_chain() -> None:
         await _run(conn, script, sent=sent, settings=settings)
 
     asyncio.run(scenario())
-    details = [card for card in sent if card.card_kind != CARD_SUMMARY]
-    assert [(card.chain_name, card.biz_key, card.sample_index) for card in details] == [
-        (CHAIN_BUY, "ORD-A", 1),
-        (CHAIN_SELL, "ORD-C", 1),
-    ]
-    status = {row["biz"]: row["status"] for row in conn.results if row["kind"] == "master_order"}
-    assert status == {
-        "ORD-A": "sent",
-        "ORD-B": "pending",
-        "ORD-C": "sent",
-        "ORD-D": "pending",
-    }
-
-    async def again() -> None:
-        script = _script()
-        script["orders"] = orders
-        script["issue"] = lambda _data: []
-        script["sales"] = lambda _data: []
-        script["trs"] = lambda _data: []
-        await _run(conn, script, sent=sent, settings=settings)
-
-    asyncio.run(again())
-    details = [card for card in sent if card.card_kind != CARD_SUMMARY]
-    assert [(card.chain_name, card.biz_key, card.sample_index) for card in details] == [
-        (CHAIN_BUY, "ORD-A", 1),
-        (CHAIN_SELL, "ORD-C", 1),
-        (CHAIN_BUY, "ORD-B", 1),
-        (CHAIN_SELL, "ORD-D", 1),
-    ]
+    assert [card.chain_name for card in sent] == [DIM_ORDER]
+    assert sent[0].sample_index is None
+    formulas = {line.formula_text for line in sent[0].lines}
+    assert {"买1", "买2", "卖1", "卖2"} <= formulas
+    assert conn.results[0]["status"] == "sent"
 
 
 def test_push_limit_above_count_sends_all() -> None:
@@ -1210,11 +1173,12 @@ def test_push_limit_above_count_sends_all() -> None:
         await _run(conn, script, sent=sent, settings=settings)
 
     asyncio.run(scenario())
-    assert [card.biz_key for card in sent if card.card_kind != CARD_SUMMARY] == ["ORD-A", "ORD-B"]
+    assert [card.chain_name for card in sent if card.card_kind == CARD_DIMENSION] == [DIM_ORDER]
+    assert {line.formula_text for line in sent[0].lines} >= {"错1", "错2"}
 
 
-def test_correct_chain_summary_is_sent_when_push_filter_is_error_only() -> None:
-    """概况卡的总结论是正确时，只推错误的配置仍然先发概况。明细保持 pending。"""
+def test_correct_dimension_is_sent_when_push_filter_is_error_only() -> None:
+    """方向卡总结论是正确时，只推错误的配置仍然发送。三个计数按这张卡。"""
 
     conn = _Own()
     sent: list[object] = []
@@ -1229,28 +1193,24 @@ def test_correct_chain_summary_is_sent_when_push_filter_is_error_only() -> None:
         await _run(conn, script, sent=sent, settings=settings)
 
     asyncio.run(scenario())
-    assert [card.card_kind for card in sent] == [CARD_SUMMARY]
-    assert sent[0].chain_name == CHAIN_BUY
-    assert [(owner.biz_key, owner.verdict) for owner in sent[0].owners] == [("ORD-A", "correct")]
-    assert sent[0].owners[0].lines == ()
-    detail = next(row for row in conn.results if row["kind"] == "master_order")
-    summary = next(row for row in conn.results if row["kind"] == CARD_SUMMARY)
-    assert detail["status"] == "pending"
-    assert summary["status"] == "sent"
-    assert summary["verdict"] == "correct"
+    assert [card.card_kind for card in sent] == [CARD_DIMENSION]
+    assert sent[0].chain_name == DIM_ORDER
+    detail = conn.results[0]
+    assert detail["status"] == "sent"
+    assert detail["verdict"] == "correct"
     assert conn.runs[0]["correct"] == 1
     assert conn.runs[0]["error"] == 0
 
 
-def test_summary_resend_keeps_owner_without_rereading_lines() -> None:
-    """概况失败、明细已发出时，重发概况仍带归属，不再读结论行。"""
+def test_dimension_resend_keeps_the_same_lines() -> None:
+    """方向卡失败后重发仍是这一张，不新插结论行。"""
 
     conn = _Own()
     seen: list[object] = []
 
     async def sender(_settings: object, card: object) -> SendResult:
         seen.append(card)
-        if card.card_kind == CARD_SUMMARY and not card.resend:
+        if not card.resend:
             return SendResult(ok=False, retryable=True, attempted=True)
         return SendResult(ok=True, retryable=False, attempted=True)
 
@@ -1265,14 +1225,15 @@ def test_summary_resend_keeps_owner_without_rereading_lines() -> None:
 
     asyncio.run(scenario())
     resent = seen[-1]
-    assert resent.card_kind == CARD_SUMMARY
+    assert resent.card_kind == CARD_DIMENSION
     assert resent.resend is True
-    assert resent.owners[0].biz_key == "ORD-A"
-    assert resent.owners[0].lines == ()
+    assert resent.chain_name == DIM_ORDER
+    assert any(line.formula_text == "只有一行" for line in resent.lines)
+    assert sum(1 for row in conn.lines if row["formula"] == "只有一行") == 1
 
 
-def test_existing_round_backfills_summary_before_pending_details() -> None:
-    """旧轮次只有明细时，下次启动补插概况并先发。三个计数不因补插增加。"""
+def test_existing_round_does_not_backfill_a_summary() -> None:
+    """已经落下的旧归属卡下次只重发自己，不再补插概况，也不重算结论行。"""
 
     conn = _Own()
     sent: list[object] = []
@@ -1290,14 +1251,13 @@ def test_existing_round_backfills_summary_before_pending_details() -> None:
         await _run(conn, script, sent=sent)
 
     asyncio.run(scenario())
-    assert [card.card_kind for card in sent] == [CARD_SUMMARY, "master_order"]
-    assert [card.record_no for card in sent] == ["2", "1"]
-    assert [(owner.biz_key, owner.verdict) for owner in sent[0].owners] == [("ORD-A", "error")]
+    assert [card.card_kind for card in sent] == ["master_order"]
+    assert sent[0].record_no == "1"
     assert len(conn.lines) == 1
     assert conn.lines[0]["formula"] == "只有一行"
     assert conn.runs[0]["error"] == 1
     assert conn.runs[0]["correct"] == 0
-    assert len([row for row in conn.results if row["kind"] == CARD_SUMMARY]) == 1
+    assert [row for row in conn.results if row["kind"] == CARD_SUMMARY] == []
 
 
 def test_completed_run_silences_unsent_snapshot() -> None:
@@ -1335,7 +1295,7 @@ def test_completed_run_silences_unsent_snapshot() -> None:
         await _run(conn, script, scan=GateScan(ready=(READY,), lines=()), sender=sender)
 
     asyncio.run(scenario())
-    assert calls == ["1", "2"]
+    assert calls == ["1"]
     assert {row["status"] for row in conn.results} == {"sent"}
     detail = next(row for row in conn.results if row["kind"] == "clearing_batch")
     assert detail["biz"] == "SB-OPEN"
@@ -1385,7 +1345,7 @@ def test_snapshot_table_change_adds_a_card() -> None:
     asyncio.run(scenario())
     assert len(conn.runs) == 1
     assert len([row for row in conn.results if row["kind"] == "clearing_batch"]) == 2
-    assert len([row for row in conn.results if row["kind"] == CARD_SUMMARY]) == 1
+    assert [row for row in conn.results if row["kind"] == CARD_SUMMARY] == []
 
 
 def test_clip_marks_overflow_and_biz_key_is_stable() -> None:

code/post_clear_check/tests/test_schema.py

@@ -2,7 +2,7 @@
 
 import re
 
-from post_clear_check.chain import CHAINS
+from post_clear_check.chain import CHAINS, DIMENSIONS
 from post_clear_check.config import tool_root
 
 BUSINESS_DATABASES = ("db_issue_clear", "db_issue", "db_sales", "db_trs")
@@ -50,7 +50,7 @@ def test_schema_creates_only_own_database() -> None:
 
 
 def test_check_result_chain_name_allows_registered_codes() -> None:
-    """chain_name 默认空串，CHECK 只允许登记表里的 12 个代码和空串。"""
+    """chain_name 允许四个方向、原来的 12 个内部链路代码和空串。"""
 
     text = _schema()
     block = text[text.index("CREATE TABLE check_result") : text.index("CREATE TABLE check_line")]
@@ -58,15 +58,15 @@ def test_check_result_chain_name_allows_registered_codes() -> None:
     constraint = block[block.index("chk_check_result_chain_name") :]
     constraint = constraint[: constraint.index(")\n  ),") + 1]
     allowed = set(re.findall(r"'([a-z_]*)'", constraint))
-    assert allowed == set(CHAINS) | {""}
+    assert allowed == set(CHAINS) | set(DIMENSIONS) | {""}
     assert len(CHAINS) == 12
 
 
-def test_check_result_card_kind_includes_chain_summary() -> None:
-    """概况卡用 chain_summary。明细的四种归属仍在。"""
+def test_check_result_card_kind_includes_dimension() -> None:
+    """方向卡用 dimension。历史归属和概况卡仍留在 CHECK 里。"""
 
     text = _schema()
     block = text[text.index("chk_check_result_card_kind") :]
     block = block[: block.index(")")]
-    for kind in ("master_order", "fund_account", "trs_user", "clearing_batch", "chain_summary"):
+    for kind in ("dimension", "master_order", "fund_account", "trs_user", "clearing_batch", "chain_summary"):
         assert f"'{kind}'" in block

doc/业务链路核对设计.md

@@ -6,15 +6,17 @@
 
 ## 1. 结论
 
-核对单位从「两列是否相等」改成「一条业务链路」。范围内一共 11 条链路。每条链路按节点往下看：这一环应不应该出现，出现了的字段算得对不对。字段公式写在节点里，不再单独成为一张卡的主题。
+核对仍按 11 条业务链路的节点往下算：这一环应不应该出现，出现了的字段算得对不对。字段公式写在节点里。飞书不再按「一条链路、一个归属对象」发卡。
 
-一张飞书卡只承载一条链路、一个归属对象。标题写结论、链路、归属和主键。正文先写「涉及」，再按段写依据、中文公式、代入、期望值与当前值。
+对外只发四个方向：订单链路、发行报表、销售报表、TRS 报表。一张方向卡汇总这一端本轮的全部结论行。标题写方向和结论。正文按摘要、分节或分报表、错误清单、可疑清单、无法展开来排。计算过程仍是字段、期望、实据、依据、结论。
+
+内部链路代码留在结论行上，用来查依据和涉及步骤。方向只决定收进哪一张卡，不改公式、人口和挂链。OTC003 的结论行仍挂在买入执行或卖出执行上，展示时按产生来源归进发行报表或销售报表，不在订单卡里再写一遍。
 
 三端 23 个清算步骤不各建一条链路。有权威公式且清算后仍能复核的步骤，挂到已有链路做节点；没有权威口径或事后无法复核的，在本文写不支持及原因，不发卡。
 
-本次实现只搭链路骨架和卡片版式，把现有结论行挂到 11 条链路上。不重写任何一条链路的节点公式。唯一的删除是增额、展期与 OTC009、OTC012 落库表的比对：这两张表没有写入方，节点本身不成立（见 4.8、4.9）。
+增额、展期不与 OTC009、OTC012 落库表比对：这两张表没有写入方，节点本身不成立（见 4.8、4.9）。
 
-以后只做两种改动：新增一条链路，或修改已有链路里的节点。不按报表编号横向改公式。
+改公式时仍先改本文第 4 节的对应链路，再改代码。不按报表编号横向改公式。
 
 ## 2. 核对范围
 
@@ -433,201 +435,121 @@ TRS 的字段公式不在 Excel 里，在产品原型 `TRS/4 业务引起数据
 
 ## 7. 飞书正文
 
-### 7.0 链路概况
+对外只发四张方向卡，顺序固定为订单链路、发行报表、销售报表、TRS 报表。没有结论行的方向不发。核对门禁没有开跑、本轮不算报表时，仍只落原来的清算批次卡，不发四张空卡。
+
+`check_result.card_kind` 为 `dimension`，`chain_name` 为 `order_chain`、`issue_report`、`sales_report`、`trs_report`，`biz_key` 为营业日。`check_run` 的三个计数是这四张卡的总结论，不含已经不再生成的链路概况。已经 `sent` 的旧归属卡不撤回。
 
-明细卡之前，每条本轮有明细卡的链路先发一张概况。没有明细的链路不发。
+四个方向不受 `POST_CLEAR_CHECK_FEISHU_PUSH` 的结论类型和条数截断。截断会直接少掉一个方向。
 
-标题：
+### 7.1 标题
 
 ```text
-链路-买入执行 正确2条，错误1条，可疑0条
+订单链路 可疑
+发行报表 正确
 ```
 
-- 「条」是这条链路下的归属个数，一张明细卡计 1，不是结论行数。一张明细卡上同时有正确行和错误行时，只计在它的总结论那一档（错误高于可疑，可疑高于正确）。
-- 顺序固定为正确、错误、可疑，数量为 0 也写。
+- 有错误为红，否则有可疑或无法展开为橙，否则为蓝。无法展开在库里仍是 `suspect`。
 - 失败重发在末尾加「重发」。
-- 标题不写系统和清算步骤。卡片颜色按这条链路的总结论：有错误为红，否则有可疑为橙，否则为蓝。
-- 概况卡不受推送配置过滤。过滤只作用于后面的明细卡。
+- 标题不写系统、清算步骤、归属和营业日。
 
-正文先写正确、错误、可疑三档。档与档之间空一行（`lark_md` 的单个换行会显示成空格）。某一档没有归属时写「无」。有归属时只列名单，不附示例解读：
+### 7.2 订单链路
 
-```text
-正确：主订单：ORD1、ORD2、ORD3；资金账户：FA1、FA2
+开头三行是营业日、交易日、批次号，用单个换行连在一起。其后按章写：摘要、每单链路、盘中公式、清算对照、关联报表、错误清单、可疑清单、无法展开。没有正确行的计算章不写。三张清单没有行时写「无」。
 
-错误：主订单：ORD9
-
-可疑：无
-```
+摘要先写人口：已完成或已结算的 ORD 主单，再按主单计正确、错误、可疑笔数。已撤主单和强平主单不进入买入、卖出公式。三端门禁各一行，只写有没有差异，公式留在对应报表卡。
 
-- 归属名称与明细卡标题相同：主订单、资金账户、用户账号、清算批次。同一归属的主键用顿号分开，不同归属用分号分开，主键按字典序。
-- 超长时先从正确名单末尾删主键，再删可疑，最后删错误。删过的那一档写成 `主订单：ORD1、ORD2…等共200个`。三档名单都保留。概况不拆成多张。
+同一字段的正确行收成一块，结论写「正确（n/n）」。同字段里有错误或可疑时，正确部分仍收成一块，错误和可疑逐条展开，不并进这块。
 
-概况卡落在 `check_result`：`card_kind` 为 `chain_summary`，`biz_key` 为链路代码，没有 `check_line`。`check_run` 的三个计数不含概况卡。已 `sent` 的概况不重发，因此不会因为这次版式单独再发；`pending` 和 `failed` 下次用新正文发。
-
-### 7.1 结构
-
-以下是明细卡。概况卡见 7.0，概况不增加章节。
+关联报表不重算。只列本轮主单在三张报表卡上的对错笔数，例如「发行 OTC003：本轮主单 2 笔，正确 2。明细在发行报表。」没有主单号的账户级报表不写进这一章。
 
 ```text
-营业日：2026-10-21
-交易日：2026-10-21
-批次号：CB20261021001
-涉及：发行端清算步骤3 订单及票据处理；销售端清算步骤3 订单及票据处理；TRS清算步骤5 订单与资金清算
+订单链路 可疑
+
+营业日：2026-10-30
+交易日：2026-10-30
+批次号：ICLR…
 
 摘要
-本卡正确 12，错误 1，可疑 1。
 
-错误清单
-……
+人口：已完成或已结算的 ORD 主单。本方向正确 2 笔，错误 0 笔，可疑 1 笔。
 
-可疑清单
-……
+已撤主单、强平主单不进入买入/卖出公式。
 
-无法展开
-……
+发行门禁：无差异。明细在发行报表。
 
-正确
-……
+盘中公式
 
-不适用
-……
-```
+字段：票据面值
+期望：执行名义本金（结算币种）× A 类报价
+实据：ORD1、ORD2 无差异
+依据：共赢票据重要字段、计算公式等说明V2 工作表「买入订单」第 24 行
+结论：正确（2/2）
 
-- 配置了推送条数时，正文第一行是「抽样明细第 n 条」。n 是这条链路本次抽样的序号，从 1 起。没配条数或配 0 时不写这行。拆成多张发送时每一张都保留这行。
-- 开头接着是营业日、交易日、批次号。
-- 其后一行是「涉及」，写这条链路经过的系统和清算步骤，用清算步骤名，不写 Python 模块路径。核对门禁和未归链路写「涉及：核对工具自身」。标成不支持的步骤不出现在这行。抽样行和这四行在删段、拆分时都不丢。
-- 其后按章写：摘要、错误清单、可疑清单、无法展开、正确、不适用。章与章之间空一行。某一章没有行时写「无」。主题为错误时只写摘要和错误清单，不写其余四章。
-- 飞书 `lark_md` 把单个换行显示成空格。六句之间、章标题和内容之间都空一行。开头四行仍用单个换行连在一起。
+关联报表
 
-摘要只数这张正文实际写上的结论行：正确、错误、可疑。无法展开算在可疑里，不适用算在正确里。库里的 `verdict` 仍只有这三档，不新增结论。
+发行 OTC003：本轮主单 2 笔，正确 2。明细在发行报表。
 
-### 7.2 章节里的六句
+错误清单
 
-错误行、期望值算得出的可疑行、期望值为空的可疑行、不适用的正确行，各自展开成六句。同一报表、同一依据（去掉列名之后）的其余正确行收成一块，不逐列展开。
+无
 
-| 行 | 内容 |
-|---|---|
-| 节点 | 中文公式。没有公式时写「核对」 |
-| 报表 | 被核对对象里的报表名。没有报表名则不写 |
-| 筛选 | 有报表名时只写归属。没有报表名时写链路名称，有归属再以「 · 」接上。归属和报表名都来自这张卡已有的被核对对象，不另查范围 |
-| 期望 | 中文公式。期望值算不出时写「期望值无法计算」 |
-| 实据 | 代入式（与公式右侧相同或整段就是公式时不写）和当前值。正确覆盖块写「n 列无差异」；只有一列时，代入式仍写在这句前面 |
-| 依据 | 三种依据之一的定位，去掉列名 |
-| 结论 | `结论：` 加「正确」「错误」或「可疑」，再接保留下来的说明和期望句 |
+可疑清单
 
-说明仍是代入式分号后面的句子。只是「同行等式」的不写。回写、口径差异留在结论句里，不再单独成行。
+节点：订单卖出股数
+期望：按卖出比例
+实据：ORD9
+依据：共赢票据重要字段、计算公式等说明V2 工作表「卖出订单」第 19 行
+结论：可疑。无法判定。
 
-期望句仍是这四种，写在结论里，不另作一档结论：
+无法展开
 
-- 正确：`期望 X，当前 X。符合预期。` 正确覆盖块不重复这句，结论只写「正确」，有说明再接上。
-- 错误：`期望 X，当前 Y。不符合预期。`
-- 字段缺失、无法代入：`期望值无法计算，当前 Y。` 这种可疑行放进「无法展开」，结论仍是可疑。
-- 期望值算得出但仍是可疑：`期望 X，当前 Y。无法判定。` 放进「可疑清单」。
+无
+```
 
-公式或说明里已经有「不出现」「不适用」「未触发」的正确行放进「不适用」，不放进正确覆盖块。没有这些字的正确行不为了分章改公式。
+章节之间空一行。一张报表或一个字段内部用单个换行，五行连在一起。空的错误、可疑、无法展开各写一行「：无」。
 
-中文名只用原型和代码里该端已有的列名。两端同编号报表的列名不完全相同，卡片写该端自己的列名。
+### 7.3 三张报表
 
-主题为错误时，正文在分章之前就只留错误行。摘要写成 `本卡正确 0，错误 n，可疑 0。正确与可疑留在核对明细。` 正确行和可疑行留在 `check_line`，不出现在这张卡的其余章节里。
+发行、销售、TRS 用同一版式：批次门禁、摘要、分报表、错误清单、可疑清单、无法展开。某一张报表没有结论行时，不编一块「正确」。
 
-### 7.3 示例
+批次门禁只收这一端的门禁行。没有门禁差异时不写这一章。摘要按报表编号计正确、错误、可疑的张数，门禁行不计入张数，但门禁错误仍让这张卡的总结论成为错误。
 
-主题为错误的资金变动卡。同一资金账户上如果还有符合预期或可疑的节点，那些行留在 `check_line`，不出现在下面这张卡里：
+同一报表、同一依据的正确列收成一块。错误和可疑逐条写节点、期望、实据、依据、结论。主题是错误时，正确列仍留在分报表，不写「正确与可疑留在核对明细」。
 
 ```text
-错误 资金变动 资金账户 FA00012345
-
-营业日：2026-10-13
-交易日：2026-10-13
-批次号：CB20261013001
-涉及：销售端清算步骤3 订单及票据处理；销售端清算步骤7 营业日日切
+销售报表 错误
 
 摘要
 
-本卡正确 0，错误 1，可疑 0。正确与可疑留在核对明细。
-
-错误清单
-
-节点：买入待交收流水金额 = −订单总金额
-
-筛选：资金变动 · 主订单 ORD001202610130004
-
-期望：买入待交收流水金额 = −订单总金额
-
-实据：主订单编号 ORD001202610130004 的总金额 220421.76。当前 -220400.00
-
-依据：共赢票据重要字段、计算公式等说明V2 工作表「买入订单」第 28 行
-
-结论：错误。清算当批新流水没有流水编号，按订单总金额对。期望 -220421.76，当前 -220400.00。不符合预期。
-```
-
-同一报表、同一依据的正确列收成一块。账户资产卡上若只剩这类正确列，正确章是：
+本方向正确 1 张报表，错误 1，可疑 0。
 
-```text
-正确
+分报表
 
 报表：OTC001 资金账户资金报表
-
-筛选：资金账户 FA00012345
-
-节点：现金余额 = 现金余额（已交收） − 现金余额（买入待交收） + 现金余额（卖出待交收）
-
-期望：与节点公式一致
-
+筛选：资金账户 FA1
+期望：现金余额 = 现金余额（已交收） − 现金余额（买入待交收） + 现金余额（卖出待交收）
 实据：1 列无差异
-
 依据：共赢票据重要字段、计算公式等说明V2 工作表「资金账户钱包信息」第 3 行
-
 结论：正确
-```
 
-期望值算不出的可疑行在「无法展开」，例如最新价缺失。结论仍是可疑，不单开第四种结论。
-
-### 7.4 超长与拆分
-
-主题为错误时，正文在计长度之前就只含错误行，摘要带「正确与可疑留在核对明细」。其余主题超过 3500 字时，从正确章末尾整块删掉覆盖块（同一报表、同一依据的一块一起删）。不适用、错误、可疑和无法展开不删。摘要的三个计数只算还写在正文里的行；删掉的正确列写成 `正确另有 k 列未写入本卡`。
-
-请求体超过约 28KB 时按结论行拆成多张。每一张按自己带上的行重新分章，错误行和可疑行不丢。每张都带抽样行、开头四行、同一个记录编号和「第几张，共几张」。章节内部的空行不单独成段。
-
-### 7.5 订单链路
-
-买入执行、卖出执行不用 7.1 的摘要和五档清单。开头四行之后按这个顺序写，没有行的章不写：
-
-```text
-每单链路
-
-结论：正确。执行事实列与源表一致。对冲记录成交股数等于 TRS 执行根的成交股数。
-
-盘中公式
-
-字段：票据面值
-期望：1000000.00 × 10% = 100000.00
-实据：执行票据面值 = 100000.00
-依据：共赢票据重要字段、计算公式等说明V2 工作表「买入订单」第 24 行
-结论：正确
+错误清单
 
-清算对照
+报表：OTC002 现金变动记录
+筛选：资金账户 FA00000001 · 主订单 ORD001
+节点：买入待交收流水金额 = −订单总金额
+期望：−220421.76
+实据：当前 -220400.00
+依据：共赢票据重要字段、计算公式等说明V2 工作表「买入订单」第 28 行
+结论：错误。期望 -220421.76，当前 -220400.00。不符合预期。
+```
 
-字段：T日订单结算与工作快照一致
-期望：与字段公式一致
-实据：work/3 与 cleared/0 一致
-依据：研发设计 清算步骤核对改造方案「5.1 发行端（6 步）」
-结论：正确
+### 7.4 超长与拆分
 
-关联报表
+方向卡不按 3500 字删掉报表块或公式段。删整章会只剩「正确 N 张」和空清单。请求体超过约 28KB 时按结论行拆成多张。每张带同一个记录编号和「第几张，共几张」。拆开发送时整卡仍只有一个 `notify_status`。前面已发出的张下次会再发一遍。
 
-字段：总金额
-期望：票据发行价 + 承销费用
-实据：OTC003 共赢票据订单报表.总金额(结算币种) = 118304.88
-依据：产品原型 OTC003
-结论：正确
-```
+### 7.5 不改的计算
 
-- 每单链路只写结论。这些行都正确时，等号写成「等于」，接在「结论：正确。」后面。有错误或可疑时，结论用最严重的一档，并且只展开这些行。
-- 盘中公式、清算对照、关联报表每个字段五行：字段、期望、实据、依据、结论。飞书里单个换行会变成空格，所以这五行之间空一行，这是能各自成行的最紧凑排法。章与章、字段与字段之间也只空这一行。
-- 正确行的结论只写「正确」。错误行写「错误。不符合预期」，可疑行写「可疑。无法判定」或「可疑。期望值无法计算」。说明仍接在结论后面。
-- 分章只看这张卡上已有的公式和被核对对象：执行事实、对冲执行根进每单链路；含「工作快照」「已发布结果」的进清算对照；带报表名的进关联报表；其余计算进盘中公式。
-- 主题为错误时仍只带错误行，开头加「正确与可疑留在核对明细。」超长时从这三章末尾删正确字段，每单链路、错误和可疑不删，并写「另有 k 项未写入本卡」。
+方向卡不新增公式，也不把巡检稿里的业务判断写成新的对错。已撤主单、强平主单、对冲比例低于建议比例、手填汇率偏离、持仓差额归因、跨表加总，都不在这四张卡里单独成行。库内结论仍只有 `correct`、`error`、`suspect`。
 
 ## 8. 这一步实现什么
 

doc/使用说明.md

@@ -75,8 +75,8 @@ POST_CLEAR_CHECK_BUSINESS_DATE=
 # 也可以写成：错误、可疑
 POST_CLEAR_CHECK_FEISHU_PUSH=
 
-# 每条链路本次最多推送的明细张数。按该链路待发送列表从前往后取。
-# 大于该链路现有张数、填 0 或留空，则这条链路全部推送。概况卡不占这个名额。
+# 旧归属卡按链路限制本次推送张数。四个方向卡不受这个上限。
+# 大于该链路现有张数、填 0 或留空，则这条链路的旧卡全部推送。
 POST_CLEAR_CHECK_FEISHU_PUSH_LIMIT=
 ```
 
@@ -86,9 +86,9 @@ POST_CLEAR_CHECK_FEISHU_PUSH_LIMIT=
 
 `FEISHU_SIGN_SECRET` 可留空。机器人开了签名校验时填密钥。`FEISHU_WEBHOOK` 留空时结论仍落库，发送状态记为 `failed`。
 
-`FEISHU_PUSH` 留空时三类都推。只写 `错误` 时，正确和可疑的卡片仍写入 `check_result`，状态保持 `pending`，不调用飞书。之后把类型加回配置，下次启动会补发这些 `pending` 卡。日志里「飞书卡片已发送」「飞书发送未成功」「飞书卡片未能发送」在未配置推送类型时会带「推送类型未配置，三类都推」。这表示没有按结论类型过滤，不是没配 Webhook。配了类型之后，被跳过的卡打「不在推送配置内」。
+`FEISHU_PUSH` 留空时三类都推。只写 `错误` 时，正确和可疑的旧归属卡、门禁快照仍写入 `check_result`，状态保持 `pending`，不调用飞书。四个方向卡不看这个过滤，正确的方向也会发。之后把类型加回配置，下次启动会补发这些 `pending` 卡。日志里「飞书卡片已发送」「飞书发送未成功」「飞书卡片未能发送」在未配置推送类型时会带「推送类型未配置，三类都推」。这表示没有按结论类型过滤，不是没配 Webhook。配了类型之后，被跳过的旧卡打「不在推送配置内」。
 
-`FEISHU_PUSH_LIMIT` 是每条链路的明细张数上限，和上面的类型过滤叠在一起。每条链路自己的待发明细按主键从前往后排，只发前 N 张。买入执行抽 1 张，不会占用卖出执行或资金变动的名额。概况卡不计数。N 大于该链路现有的待发张数、填 0 或留空时，这条链路的待发明细全部发送。配了 N 时，明细卡正文第一行写「抽样明细第 n 条」，n 是这条链路本次抽样的序号，从 1 起。超出的卡保持 `pending`，下次启动再从该链路剩下的最前面取 N 张。
+`FEISHU_PUSH_LIMIT` 是旧归属卡按链路计数的张数上限，和上面的类型过滤叠在一起。四个方向卡不受这个上限，否则会少发一整个方向。旧归属卡仍按主键从前往后排，每条链路只发前 N 张。N 大于该链路现有的待发张数、填 0 或留空时，这条链路的待发明细全部发送。配了 N 时，被抽中的旧明细卡正文第一行写「抽样明细第 n 条」。超出的卡保持 `pending`。
 
 路径本身用环境变量指出，不要写进配置文件里：
 
@@ -127,7 +127,7 @@ python -m post_clear_check
 | `飞书卡片不推送，…不在推送配置内` | 这张卡的结论不在 `FEISHU_PUSH` 里，状态仍是 `pending` |
 | `飞书卡片超出推送条数，…` | 这条链路本次已发满 `FEISHU_PUSH_LIMIT` 张。这张没发，状态仍是 `pending`，下次启动再从该链路剩余列表前面取 |
 | `有对冲，TRS 批次未完成，发行和销售本轮不核对` | 这一天先等 TRS。核对日早于今天时另落一张错误卡 |
-| `开始<步骤>` / `<步骤>完成，用时 N 秒，…` | 每一步的开始和结束。步骤依次是检查三端清算批次、从业务库取数、核对订单和三端报表、写入自有库、发送飞书卡片。完成句后面带查询条数、明细张数、概况张数、结论行数或发送结果 |
+| `开始<步骤>` / `<步骤>完成，用时 N 秒，…` | 每一步的开始和结束。步骤依次是检查三端清算批次、从业务库取数、核对订单和三端报表、写入自有库、发送飞书卡片。完成句后面带查询条数、方向张数、结论行数或发送结果 |
 | `仍在<步骤>，已用 N 秒，…` | 不在终端里运行时（cron、重定向到文件），长步骤每 10 秒打一条，带进度和当前在读哪张表，表示程序没有卡住 |
 | `<步骤>中断，已用 N 秒` | 这一步抛错退出，原因看紧接着的错误日志 |
 
@@ -149,11 +149,11 @@ python -m post_clear_check
    - OTC001 的钱包比对和现金可用读 T 日已发布候选派生出的钱包（`kind=cleared`，当日买卖先置 0，再算现金余额、冻结和可用）。科目都是正数；报表买入待交收展示为负值，比较和代入公式时取相反数。没有已发布候选或缺本币种行时这几条记可疑。账户资产、资金变动、资金交收的步骤节点另读销售 `kind=input` 与 `kind=cleared`：第 3、4 步用 input 重算后与 cleared 比，不读 `pre_clearing`。input 里已经有该订单 posted 的解冻凭证时，不再把交易冻结扣第二次。重放 input 流水按主键 id 升序，不按查询返回的幂等哈希顺序。发行和销售日报只读 `status=active`，同一营业日的 superseded、deprecated 旧行不核对。等值港币资产净值在乘、除四舍五入到分后结果相同时，方向视为已确定。钱包只重算本批 `input` 与 `cleared` 的 T 日结果，不读实时钱包。发行交收台账读 `work/4`，`cleared/0` 应与它一致，不读实时台账。日终 `daycut_fund_eod` 只在核对门禁上与这份候选比，不一致才记错误。
    - 同一 IN 列表超过 500 个值时分几条语句读。
    - OTC007、OTC008 行按票据号挂到订单上，只核业务类型（`buy_note`、`sell_note`）与主订单方向一致、交易日相同的行。同一票据另一张订单的行不算进来；哪张已读主单都对不上的行按「有行但无法对应主订单编号」记可疑。
-6. 结论行在调用核对函数的地方挂到业务链路（见下表），收成卡片写入自有库，再发给飞书。发送时先发每条有明细的链路的概况，再发明细。一张明细卡上有多条时，总结论取最严重的一档：`error` 高于 `suspect`，`suspect` 高于 `correct`。飞书上显示为错误、可疑、正确。主题为错误的明细卡，正文只写错误清单，摘要写明正确与可疑留在核对明细；同一张卡上的正确行和可疑行仍留在 `check_line`，不出现在这张飞书卡的正文里。`check_run` 上的三个计数是明细卡片张数，不是结论行数，也不含概况卡。
+6. 结论行在调用核对函数的地方挂到业务链路（见下表），再按产生来源收成最多四张方向卡写入自有库，发给飞书。顺序是订单链路、发行报表、销售报表、TRS 报表。没有结论行的方向不发。一张卡上有多条时，总结论取最严重的一档：`error` 高于 `suspect`，`suspect` 高于 `correct`。方向卡主题是错误时，正确行仍写在正文里。`check_run` 上的三个计数是方向卡张数，不是结论行数。门禁没开跑时仍只落清算批次卡，不发四张空卡。
 
-一张卡只承载一条链路、一个归属对象，按「链路 + 归属 + 主键」分卡。同一主订单上有几条链路就是几张卡。归属按链路登记的优先顺序取结论行上第一个非空的键，不从正文反推。链路的口径和节点以 `doc/业务链路核对设计.md` 为准。
+方向卡的 `card_kind` 是 `dimension`，`chain_name` 是 `order_chain`、`issue_report`、`sales_report`、`trs_report`，`biz_key` 是营业日。结论行上的内部链路代码不改。没挂链或归属键为空的行仍是未归错误，放进产生它的那个方向。发行门禁写进发行报表的批次门禁，销售、TRS 同理。订单卡的摘要只引用三端门禁有没有差异。链路的口径和节点以 `doc/业务链路核对设计.md` 为准。
 
-| 链路 | `chain_name` | 归属（`card_kind`） | 现有结论行 |
+| 链路 | 结论行上的 `chain_name` | 结论行归属 | 现有结论行 |
 |---|---|---|---|
 | 买入执行 | `buy_execution` | 主订单 | 买入方向的订单规则、对冲执行根、两端 OTC003 |
 | 卖出执行 | `sell_execution` | 主订单 | 卖出方向（含强平、到期）的订单规则、对冲执行根、两端 OTC003 |
@@ -165,8 +165,8 @@ python -m post_clear_check
 | 展期 | `extension` | 资金账户 | 发行 `input` 展期主键集合，以及 `work/3` 的展期后到期日 |
 | 委托执行 | `trade_execution` | 主订单，对不上时用户账号 | TRS002、TRS003、TRS004、TRS005、TRS008；封存成交、T 日委托、持仓行、候选成交 |
 | 用户资产 | `user_asset` | 用户账号 | TRS001、TRS006、TRS007、TRS009、TRS010、TRS012、TRS013、TRS014、TRS016；封存资金、封存汇率、T 日债务、汇率调整、资金账户、候选资金 |
-| 核对门禁 | `check_gate` | 清算批次 | 报表早于批次、TRS 未完成、业务表缺失（1146）、发行基线与上游版本、发行发布唯一、销售锁定的发行版本、销售发布成功、TRS 日期推进、TRS 版本切换、日终快照与候选不一致。批次 completed 是开跑条件，不单独成卡 |
-| 未归链路 | `unassigned` | 清算批次 | 挂不上链路、主订单方向取不到、取数缺口、链路要求的归属键为空的行，结论一律为错误 |
+| 核对门禁 | `check_gate` | 开跑前仍是清算批次卡；开跑后并进对应报表卡 | 报表早于批次、TRS 未完成、业务表缺失（1146）、发行基线与上游版本、发行发布唯一、销售锁定的发行版本、销售发布成功、TRS 日期推进、TRS 版本切换、日终快照与候选不一致。批次 completed 是开跑条件，不单独成卡 |
+| 未归链路 | `unassigned` | 并进产生它的方向卡 | 挂不上链路、主订单方向取不到、取数缺口、链路要求的归属键为空的行，结论一律为错误 |
 
 票据存续、资金交收的报表行只有资金账户列。这两条链路的结论行取报表行的 `fund_account_no`；没有报表行时，票据存续、增额、展期取子票据或展期任务上的 `fund_account_ref`，其余取主单上的销售侧资金账户；都没有就进未归链路。资金变动、资金交收的行对得上主订单时，正文仍写主订单编号，归属仍是资金账户。
 
@@ -174,48 +174,40 @@ python -m post_clear_check
 
 OTC009、OTC012 是实时导出，没有落库结果，`report_topup_payment`、`report_extension_fee` 两张表没有写入方。增额、展期不比对这两张表，取数也不读它们。增额核的是本批 `input` 里归属本营业日的 `top_up` 主键，以及这些发生额记入 `work/3` 后的增额；展期核的是已通过且审批归属本营业日的任务主键，以及 `work/3` 的到期日。卖出释放和展期费流水不进这两条链路。
 
-飞书先发链路概况，再发明细。概况一张链路一张，标题形如 `链路-买入执行 正确2条，错误1条，可疑0条`。这里的「条」是归属个数（一张明细卡计 1），不是结论行数。一张明细卡上又有正确行又有错误行时，只出现在它的总结论那一档。没有的那一档也写 0。失败重发在末尾加「重发」。标题不写系统和清算步骤。卡片颜色仍按这条链路有没有错误、可疑：有错误为红，否则有可疑为橙，否则为蓝。
+飞书按订单链路、发行报表、销售报表、TRS 报表的顺序发。标题形如 `订单链路 可疑`、`发行报表 正确`。失败重发在末尾加「重发」。标题不写系统和清算步骤。颜色按这张卡的总结论：有错误为红，否则有可疑为橙，否则为蓝。
 
-概况正文按正确、错误、可疑三档写，档与档之间空一行。某一档没有归属时写「无」。有归属时只列名单，不附示例解读：
+订单链路正文是摘要、每单链路、盘中公式、清算对照、关联报表、错误清单、可疑清单、无法展开。摘要里的笔数按主单计。同一字段的正确行收成一块，结论写「正确（n/n）」。关联报表只写本轮主单在三张报表卡上的对错笔数，不把报表列再算一遍。三端门禁在摘要里各一行。
 
-```text
-正确：主订单：ORD1、ORD2、ORD3；资金账户：FA1、FA2
+三张报表正文是批次门禁、摘要、分报表、错误清单、可疑清单、无法展开。没有门禁行时不写批次门禁。同一报表、同一依据的正确列收成一块。错误和可疑逐条写节点、期望、实据、依据、结论。方向卡是错误主题时，正确列仍留在分报表。
 
-错误：主订单：ORD9
+章节之间空一行。一张报表或一个字段内部用单个换行。空清单写成「错误清单：无」。方向卡不按 3500 字删报表或公式段。请求体超过约 28KB 时按结论行拆成多张，每张带同一个记录编号和「第几张，共几张」，整卡仍是一个 `notify_status`。公式里的制表符会去掉，`<` `>` 会改成全角。发送失败时日志带上飞书返回的 `msg`。
 
-可疑：无
-```
-
-归属名称与明细卡标题相同：主订单、资金账户、用户账号、清算批次。同一归属的主键用顿号分开，不同归属用分号分开，主键按字典序。
-
-超长时先从正确名单末尾删主键，再删可疑，最后删错误。删过的名单写成 `主订单：ORD1、ORD2…等共200个`。概况不拆成多张，也不看 `FEISHU_PUSH`。推送配置只过滤后面的明细卡。已经 `sent` 的概况不因这次版式重发；`pending` 和 `failed` 下次启动用新正文发。
-
-概况行写在 `check_result`，`card_kind` 为 `chain_summary`，`biz_key` 是链路代码，没有结论行。已经 `sent` 的概况不重发。旧轮次还没有概况行时，下次启动补插后再发，不重算明细、不新插结论行。
-
-已有库要先执行下面两句，再启动。不要重跑 `schema.sql`，那会清空已有轮次。没执行时概况行插不进去：新的一轮会整段回滚；已经落下的轮次这次发不出去，库里的明细还在。
+已有库要先放宽两道 CHECK，再启动。不要重跑 `schema.sql`，那会清空已有轮次。没执行时新的方向卡插不进去，这一轮会整段回滚。
 
 ```sql
 ALTER TABLE check_result DROP CHECK chk_check_result_card_kind;
 ALTER TABLE check_result ADD CONSTRAINT chk_check_result_card_kind CHECK (
   card_kind COLLATE utf8mb4_bin IN (
-    'master_order', 'fund_account', 'trs_user', 'clearing_batch', 'chain_summary'
+    'dimension', 'master_order', 'fund_account', 'trs_user', 'clearing_batch', 'chain_summary'
+  )
+);
+ALTER TABLE check_result DROP CHECK chk_check_result_chain_name;
+ALTER TABLE check_result ADD CONSTRAINT chk_check_result_chain_name CHECK (
+  chain_name COLLATE utf8mb4_bin IN (
+    '',
+    'order_chain', 'issue_report', 'sales_report', 'trs_report',
+    'buy_execution', 'sell_execution', 'cash_change', 'cash_settlement',
+    'account_asset', 'note_holding', 'topup', 'extension',
+    'trade_execution', 'user_asset', 'check_gate', 'unassigned'
   )
 );
 ```
 
-明细卡标题一行，顺序是结论、链路名称、归属名称、主键值，形如 `错误 买入执行 主订单 ORD001202610210004`、`错误 未归链路 清算批次 2026-10-21`。失败重发在末尾加「重发」。标题不写系统和清算步骤。
-
-正文在抽样行之后是营业日、交易日、批次号和「涉及」。「涉及」写这条链路经过的系统和清算步骤。核对门禁写「核对工具自身」，并写已经挂上的清算步骤。未归链路只写「核对工具自身」。配了推送条数时，正文第一行是「抽样明细第 n 条」。其后按章写：摘要、错误清单、可疑清单、无法展开、正确、不适用。摘要只数这张正文里的行。错误行、期望值算得出的可疑行、期望值为空的可疑行（无法展开）、公式或说明里带「不出现 / 不适用 / 未触发」的正确行（不适用），各自写成节点、报表、筛选、期望、实据、依据、结论。同一报表、同一依据的其余正确行收成一块，实据写「n 列无差异」。筛选只用来自被核对对象的报表名和归属，没有报表名时写链路名称。飞书 `lark_md` 把单个换行显示成空格，章与章、六句与六句之间空一行。期望句仍是 `期望 X，当前 Y。符合预期。`、`不符合预期。` 或 `无法判定。`，写在结论里。期望值算不出时写「期望值无法计算」，结论仍是可疑，放在无法展开。说明只保留回写、口径差异，不重复「同行等式」，接在结论里。
-
-买入执行和卖出执行不用上面的清单，改成四章：每单链路、盘中公式、清算对照、关联报表。没有行的章不写。每单链路只写一句结论。另外三章每个字段是字段、期望、实据、依据、结论，五行之间空一行。正确行的结论只写「正确」。
-
-主题为错误时，正文在计长度之前就只含错误行，摘要加「正确与可疑留在核对明细」，不写可疑、无法展开、正确、不适用四章。买入执行和卖出执行同样只含错误行，这句话写在四章前面。其余主题超过 3500 字时，从正确章末尾整块删掉覆盖块，不适用、错误、可疑和无法展开保留；摘要另写「正确另有 k 列未写入本卡」。订单链路则从盘中公式、清算对照、关联报表的末尾删正确字段，写「另有 k 项未写入本卡」。开头四行不删，并附上 `check_result.id` 作为记录编号。请求体按飞书 30KB 上限按结论行拆成多张，每张按自己的行重新分章，都带开头四行、同一记录编号和「第几张，共几张」。公式里的制表符会去掉，`<` `>` 会改成全角，避免卡片内容被拒。发送失败时日志带上飞书返回的 `msg`。
-
-同一轮次由唯一键识别：市场、交易日、三端批次 id、三端 `end_time`、规则版本。规则版本是订单和三端报表规则源文件、列清单 `columns.py`、取数 `reader.py`、销售资金步骤 `fund_chain.py`、链路登记表 `chain.py` 的 sha256。卡片幂等键在卡片种类后面加链路代码，同一主订单的两条链路是两张卡、两个幂等键。这些文件有一个字节变化，就新开一轮，旧轮次保留。第一次换成这个版本时，已经核对过的营业日会再跑一遍并发卡。命中已有轮次时不重新取数、不新插结论行；缺概况卡时只补插概况卡。`notify_status` 为 `pending` 或 `failed` 的卡片会重发：概况卡不看推送配置，排在前面，概况之间按链路登记顺序；明细卡仍要求结论在推送配置内，按主键升序。成功改为 `sent`。从 `failed` 再发时，标题末尾加「重发」。发行的 `applied_revision` 会记在 `check_run` 上，销售和 TRS 记 0。修订号不单独构成新的一轮；三端 `end_time` 或批次 id 没变、规则文件也没变时，再次启动仍走重发。
+同一轮次由唯一键识别：市场、交易日、三端批次 id、三端 `end_time`、规则版本。规则版本是订单和三端报表规则源文件、列清单 `columns.py`、取数 `reader.py`、销售资金步骤 `fund_chain.py`、链路登记表 `chain.py` 的 sha256。卡片幂等键在卡片种类后面加方向代码。同一营业日的四个方向是四张卡、四个幂等键。这些文件有一个字节变化，就新开一轮，旧轮次保留。第一次换成这个版本时，已经核对过的营业日会再跑一遍并发出四张方向卡。已经 `sent` 的旧归属卡不撤回。命中已有轮次时不重新取数、不新插结论行。`notify_status` 为 `pending` 或 `failed` 的卡片会重发：四个方向按固定顺序，不看推送类型和条数；旧归属卡和门禁快照仍按推送配置过滤。成功改为 `sent`。从 `failed` 再发时，标题末尾加「重发」。发行的 `applied_revision` 会记在 `check_run` 上，销售和 TRS 记 0。修订号不单独构成新的一轮；三端 `end_time` 或批次 id 没变、规则文件也没变时，再次启动仍走重发。
 
 `check_run.create_time` 和 `check_result` 的时间用上海时区的墙钟，不跟容器的 UTC。
 
-业务表不存在（MySQL 1146）时落一张核对门禁错误卡，公式是「业务表应存在」。挂不上链路、主订单方向取不到、链路要求的归属键为空的结论行落未归链路卡，`biz_key` 是营业日，结论为错误，公式以「结论行没有归属」开头，原因、原结论和原依据写在代入式后面。原因写「取数缺口：报表引用的主订单不在已读取范围」的，是报表行引用的主订单补读后仍没有主单，属于核对工具的取数问题，先查取数范围，不要当成业务数据错误；「主订单方向取不到」才是读到了主单但方向不是买卖。未知列（1054）仍然让进程失败退出。
+业务表不存在（MySQL 1146）时，在对应报表方向卡上记一条错误，公式是「业务表应存在」。开跑前的门禁快照仍单独落清算批次卡。挂不上链路、主订单方向取不到、链路要求的归属键为空的结论行并进产生它的方向卡，结论为错误，公式以「结论行没有归属」开头，原因、原结论和原依据写在代入式后面。原因写「取数缺口：报表引用的主订单不在已读取范围」的，是报表行引用的主订单补读后仍没有主单，属于核对工具的取数问题，先查取数范围，不要当成业务数据错误；「主订单方向取不到」才是读到了主单但方向不是买卖。未知列（1054）仍然让进程失败退出。
 
 正文超过列宽时保留前缀，并以「…已截断」结尾。数值仍在 `calc_value`、`check_value`。