post_clear_check/code/post_clear_check/tests/test_feishu.py

"""飞书卡片：颜色、签名、日志、重试和超长正文。"""

import asyncio
import base64
import hashlib
import hmac
import json
from dataclasses import dataclass
from datetime import date
from decimal import Decimal

import httpx
import pytest
from loguru import logger
from pydantic import SecretStr

from post_clear_check.chain import (
    CLEAR_MARK,
    CARD_DIMENSION,
    CHAIN_BUY,
    CHAIN_CASH_CHANGE,
    CHAIN_GATE,
    CHAIN_SELL,
    CHAIN_TRADE,
    CHAIN_UNASSIGNED,
    CHAINS,
    DIM_ORDER,
    DIM_SALES,
)
from post_clear_check.config import Settings
from post_clear_check.feishu import (
    BODY_LIMIT,
    CARD_SUMMARY,
    DEFAULT_TIMEOUT_SECONDS,
    FeishuError,
    OutboundCard,
    RateGate,
    SummaryOwner,
    build_payload,
    card_verdict,
    classify_response,
    expectation,
    pack_bodies,
    render_body,
    render_summary_body,
    render_summary_title,
    render_title,
    send_card,
    send_summary,
    sign_payload,
    summary_verdict,
)
from post_clear_check.order_decimal import VERDICT_CORRECT, VERDICT_ERROR, VERDICT_SUSPECT
from .support import FEISHU_SIGN_SECRET, FEISHU_WEBHOOK, sample_settings

DAY = date(2026, 10, 15)
ORDER_NO = "ORD001202610150006"
WEBHOOK = "https://hook.invalid/marker"
BUY_INVOLVED = CHAINS[CHAIN_BUY].involved()
CASH_INVOLVED = CHAINS[CHAIN_CASH_CHANGE].involved()


@dataclass(frozen=True)
class Line:
    formula_text: str
    calc_text: str
    calc_value: Decimal | None
    check_text: str
    check_value: Decimal | None
    basis: str
    verdict: str


def _line(
    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 = "依据",
) -> Line:
    return Line(
        formula_text=formula,
        calc_text=calc,
        calc_value=calc_value,
        check_text=check,
        check_value=check_value,
        basis=basis,
        verdict=verdict,
    )


def _card(
    lines: tuple[Line, ...] | list[Line],
    *,
    kind: str = "master_order",
    chain_name: str = CHAIN_BUY,
    biz_key: str = ORDER_NO,
    record_no: str = "81",
    batch_no: str = "B001",
) -> OutboundCard:
    return OutboundCard(
        card_kind=kind,
        chain_name=chain_name,
        biz_key=biz_key,
        business_date=DAY,
        trade_date=DAY,
        batch_no=batch_no,
        lines=tuple(lines),
        record_no=record_no,
    )


def _settings(*, webhook: str = WEBHOOK, secret: str = "") -> Settings:
    base = sample_settings()
    return base.model_copy(
        update={
            "feishu_webhook": SecretStr(webhook),
            "feishu_sign_secret": SecretStr(secret),
        }
    )


def _gate() -> RateGate:
    """每个用例一份配额，避免连着发把进程级窗口打满。"""

    return RateGate()


def _transport(handler) -> httpx.MockTransport:
    return httpx.MockTransport(handler)


def test_template_color_follows_card_verdict() -> None:
    """三种总结论分别用蓝、橙、红。混有错误时标题和颜色都是错误。"""

    samples = (
        (_line("公式甲", VERDICT_CORRECT), "blue", "正确"),
        (_line("公式乙", VERDICT_SUSPECT), "orange", "可疑"),
        (_line("公式丙", VERDICT_ERROR), "red", "错误"),
    )
    for line, color, label in samples:
        verdict = card_verdict((line,))
        title = render_title(verdict, CHAIN_BUY, "master_order", ORDER_NO)
        payload = build_payload(title=title, body="正文", verdict=verdict)
        assert payload["card"]["header"]["template"] == color
        assert title == f"{label} 买入执行 主订单 {ORDER_NO}"
        assert "correct" not in title
        assert "error" not in title
        assert "suspect" not in title

    mixed = card_verdict(
        (
            _line("对", VERDICT_CORRECT),
            _line("疑", VERDICT_SUSPECT),
            _line("错", VERDICT_ERROR),
        )
    )
    assert mixed == VERDICT_ERROR
    assert TEMPLATE_COLOR(mixed) == "red"
    suspect_card = card_verdict((_line("对", VERDICT_CORRECT), _line("疑", VERDICT_SUSPECT)))
    assert suspect_card == VERDICT_SUSPECT
    assert TEMPLATE_COLOR(suspect_card) == "orange"


def TEMPLATE_COLOR(verdict: str) -> str:
    return build_payload(title="标题", body="正文", verdict=verdict)["card"]["header"]["template"]


def test_sign_is_stable_hmac_of_empty_message() -> None:
    """签名字符串是时间戳、换行、密钥。消息为空字节，重复计算不变。"""

    timestamp = "1599360473"
    secret = "test-secret"
    first = sign_payload(timestamp, secret)
    second = sign_payload(timestamp, secret)
    key = f"{timestamp}\n{secret}".encode("utf-8")
    expected = base64.b64encode(hmac.new(key, b"", hashlib.sha256).digest()).decode("ascii")
    assert first == second == expected
    assert sign_payload("1599360474", secret) != first


def test_log_omits_webhook_and_secret() -> None:
    """网络异常的原文里即使带了地址和密钥，日志里也不能留下。"""

    lines: list[str] = []
    handle = logger.add(lambda message: lines.append(str(message)), diagnose=False, backtrace=False)

    def handler(_request: httpx.Request) -> httpx.Response:
        raise httpx.ConnectError(f"连接失败 {WEBHOOK} {FEISHU_SIGN_SECRET}")

    async def run() -> None:
        result = await send_card(
            _settings(webhook=WEBHOOK, secret=FEISHU_SIGN_SECRET),
            _card((_line("公式", VERDICT_ERROR),)),
            transport=_transport(handler),
            gate=_gate(),
        )
        assert result.ok is False
        assert result.retryable is True
        assert result.attempted is True

    try:
        asyncio.run(run())
    finally:
        logger.remove(handle)
    text = "".join(lines)
    assert WEBHOOK not in text
    assert "hook.invalid" not in text
    assert FEISHU_SIGN_SECRET not in text
    assert "https://" not in text
    assert "http://" not in text
    assert "连接失败" in text
    assert "可重试" in text


def test_http_429_is_retryable() -> None:
    """429 可以再发。签名业务码即使写在 429 的正文里，也仍然可重试。"""

    def handler(_request: httpx.Request) -> httpx.Response:
        return httpx.Response(429, json={"code": 19021})

    result = asyncio.run(
        send_card(
            _settings(),
            _card((_line("公式", VERDICT_ERROR),)),
            transport=_transport(handler),
            gate=_gate(),
        )
    )
    assert result.ok is False
    assert result.retryable is True
    assert result.attempted is True


@pytest.mark.parametrize("code", [19021, 19022, 19024, 999])
def test_business_code_is_not_retryable(code: int) -> None:
    """签名失败和其他非 0 业务码都不重试。"""

    ok, retryable, found = classify_response(200, json.dumps({"code": code}).encode())
    assert ok is False
    assert retryable is False
    assert found == code


def test_status_code_zero_is_success() -> None:
    """没有 code 时，StatusCode 为 0 也算成功。"""

    ok, retryable, found = classify_response(
        200,
        json.dumps({"StatusCode": 0, "StatusMessage": "success"}).encode(),
    )
    assert (ok, retryable, found) == (True, False, 0)
    ok, retryable, found = classify_response(200, json.dumps({"code": 0, "msg": "success"}).encode())
    assert (ok, retryable, found) == (True, False, 0)


def test_http_5xx_is_retryable_and_4xx_is_not() -> None:
    """5xx 可重试。400 不是网络失败，也不在可重试的状态里。"""

    assert classify_response(503, b"")[1] is True
    assert classify_response(500, b"upstream")[1] is True
    ok, retryable, _found = classify_response(400, json.dumps({"code": 0}).encode())
    assert ok is False
    assert retryable is False


def test_long_body_keeps_every_error() -> None:
    """正确覆盖块从末尾整块删。错误、可疑和开头四行留下，并写上记录编号。"""

    head = Line(
        formula_text="头正确",
        calc_text="1 = 1",
        calc_value=Decimal("1"),
        check_text="列 = 1",
        check_value=Decimal("1"),
        basis="依据甲",
        verdict=VERDICT_CORRECT,
    )
    first_error = _line("错误甲", VERDICT_ERROR)
    long_correct = Line(
        formula_text="长正确" + "测" * 4000,
        calc_text="1 = 1",
        calc_value=Decimal("1"),
        check_text="列 = 1",
        check_value=Decimal("1"),
        basis="依据乙",
        verdict=VERDICT_CORRECT,
    )
    tail = Line(
        formula_text="尾正确",
        calc_text="1 = 1",
        calc_value=Decimal("1"),
        check_text="列 = 1",
        check_value=Decimal("1"),
        basis="依据乙",
        verdict=VERDICT_CORRECT,
    )
    second_error = _line("错误乙", VERDICT_ERROR)
    suspect = _line("可疑甲", VERDICT_SUSPECT)
    body = render_body(
        business_date=DAY,
        trade_date=DAY,
        batch_no="B001",
        chain_name=CHAIN_BUY,
        lines=(head, first_error, long_correct, tail, second_error, suspect),
        record_no="81",
    )
    assert len(body) <= BODY_LIMIT
    assert "错误甲" in body
    assert "错误乙" in body
    assert "可疑甲" in body
    assert "头正确" in body
    assert "尾正确" not in body
    assert "长正确" not in body
    assert "另有 2 项未写入本卡。" in body
    assert "记录编号：81" in body
    # 删块以覆盖块为单位，开头四行（含「涉及」）保留。
    assert body.startswith(f"营业日：2026-10-15\n交易日：2026-10-15\n批次号：B001\n涉及：{BUY_INVOLVED}\n\n")
    assert "不符合预期。" in body
    assert "无法判定。" in body
    assert "结论：正确" in body

    huge_errors = tuple(_line(f"错误{index}-" + "错" * 500, VERDICT_ERROR) for index in range(10))
    huge = render_body(
        business_date=DAY,
        trade_date=DAY,
        batch_no="B001",
        chain_name=CHAIN_BUY,
        lines=huge_errors,
        record_no="82",
    )
    assert len(huge) > BODY_LIMIT
    for index in range(10):
        assert f"错误{index}-" in huge
    assert "记录编号：82" in huge


def test_short_body_matches_template_without_record_no() -> None:
    """订单卡没超长时不写记录编号。报表行进关联报表，五行之间空一行。"""

    line = Line(
        formula_text="总金额 = 票据发行价 + 承销费用",
        calc_text="1001000.00 + 0.00 = 1001000.00；同一主订单的执行值",
        calc_value=Decimal("1001000.000000"),
        check_text="OTC003 共赢票据订单报表.总金额(结算币种) = 1001000.00",
        check_value=Decimal("1001000.000000"),
        basis="共赢票据重要字段、计算公式等说明V2 工作表「买入订单」第 28 行「总金额」",
        verdict=VERDICT_CORRECT,
    )
    body = render_body(
        business_date=DAY,
        trade_date=DAY,
        batch_no="B001",
        chain_name=CHAIN_BUY,
        lines=(line,),
        record_no="81",
    )
    assert body == (
        "营业日：2026-10-15\n"
        "交易日：2026-10-15\n"
        "批次号：B001\n"
        f"涉及：{BUY_INVOLVED}\n"
        "\n"
        "关联报表\n"
        "\n"
        "字段：总金额\n"
        "\n"
        "期望：1001000.00 + 0.00 = 1001000.00\n"
        "\n"
        "实据：OTC003 共赢票据订单报表.总金额(结算币种) = 1001000.00\n"
        "\n"
        "依据：共赢票据重要字段、计算公式等说明V2 工作表「买入订单」第 28 行\n"
        "\n"
        "结论：正确。同一主订单的执行值。"
    )
    assert "记录编号" not in body


def test_gate_and_unassigned_involve_the_tool_itself() -> None:
    """核对门禁和未归链路都写核对工具自身。门禁另写已经挂上的步骤，未归链路不写步骤。"""

    gate = render_body(
        business_date=DAY,
        trade_date=DAY,
        batch_no="B001",
        chain_name=CHAIN_GATE,
        lines=(_line("公式", VERDICT_ERROR),),
        record_no="81",
    )
    assert "\n涉及：核对工具自身；" in gate
    assert "发行端清算步骤1 开启清算" in gate
    unassigned = render_body(
        business_date=DAY,
        trade_date=DAY,
        batch_no="B001",
        chain_name=CHAIN_UNASSIGNED,
        lines=(_line("公式", VERDICT_ERROR),),
        record_no="81",
    )
    assert "\n涉及：核对工具自身\n" in unassigned
    assert "清算步骤" not in unassigned


def test_fourth_line_sentences() -> None:
    """第四行四种句式：正确、错误、期望值无法计算、只看出现与否。可疑不下符合与否的结论。"""

    correct = _line("甲", VERDICT_CORRECT, calc_value=Decimal("779578.24"), check_value=Decimal("779578.240000"))
    assert expectation(correct) == "期望 779578.24，当前 779578.24。符合预期。"
    wrong = _line("乙", VERDICT_ERROR, calc_value=Decimal("-220421.76"), check_value=Decimal("-220400"))
    assert expectation(wrong) == "期望 -220421.76，当前 -220400.00。不符合预期。"
    blind = _line("丙", VERDICT_SUSPECT, calc_value=None, check_value=Decimal("580"))
    assert expectation(blind) == "期望值无法计算，当前 580.00。"
    present = _line("丁", VERDICT_CORRECT, calc_value=None, check_value=None)
    assert expectation(present) == "期望与当前一致。符合预期。"
    absent = _line("戊", VERDICT_ERROR, calc_value=Decimal("5"), check_value=None)
    assert expectation(absent) == "期望 5.00，当前缺失。不符合预期。"
    undecided = _line("己", VERDICT_SUSPECT, calc_value=Decimal("1.234567"), check_value=Decimal("1.2346"))
    assert expectation(undecided) == "期望 1.234567，当前 1.2346。无法判定。"
    nothing = _line("庚", VERDICT_SUSPECT, calc_value=None, check_value=None)
    assert expectation(nothing) == "期望值无法计算，当前缺失。"


def test_chinese_under_character_limit_is_not_cut() -> None:
    """上限按字符数，不按 UTF-8 字节数。"""

    formula = "正" * 2000
    body = render_body(
        business_date=DAY,
        trade_date=DAY,
        batch_no="B001",
        chain_name=CHAIN_BUY,
        lines=(_line(formula, VERDICT_CORRECT),),
        record_no="81",
    )
    assert formula in body
    assert "记录编号" not in body
    assert len(body) <= BODY_LIMIT
    assert len(body.encode("utf-8")) > BODY_LIMIT


def test_title_is_verdict_chain_owner_and_key() -> None:
    """标题顺序是结论、链路名称、归属名称、主键值。不写系统和清算步骤，失败重发末尾加「重发」。"""

    assert render_title(VERDICT_ERROR, CHAIN_BUY, "master_order", "ORD001202610210004") == (
        "错误 买入执行 主订单 ORD001202610210004"
    )
    assert render_title(VERDICT_CORRECT, CHAIN_CASH_CHANGE, "fund_account", "FA00012345") == (
        "正确 资金变动 资金账户 FA00012345"
    )
    assert render_title(VERDICT_SUSPECT, CHAIN_TRADE, "trs_user", "U0001") == "可疑 委托执行 用户账号 U0001"
    assert render_title(VERDICT_ERROR, CHAIN_GATE, "clearing_batch", "CB20261021001") == (
        "错误 核对门禁 清算批次 CB20261021001"
    )
    assert render_title(VERDICT_ERROR, CHAIN_UNASSIGNED, "clearing_batch", "2026-10-21") == (
        "错误 未归链路 清算批次 2026-10-21"
    )
    assert render_title(VERDICT_ERROR, CHAIN_BUY, "master_order", "ORD001202610210004", resend=True) == (
        "错误 买入执行 主订单 ORD001202610210004 重发"
    )
    assert "重发" not in render_title(VERDICT_ERROR, CHAIN_BUY, "master_order", ORDER_NO)
    for title in (render_title(VERDICT_ERROR, CHAIN_BUY, "master_order", ORDER_NO),):
        assert "清算步骤" not in title
        assert "发行端" not in title
    with pytest.raises(FeishuError, match="链路"):
        render_title(VERDICT_ERROR, "", "master_order", ORDER_NO)


def test_redirect_is_not_followed() -> None:
    """302 只看到一次请求，并且不可重试。"""

    calls: list[str] = []

    def handler(request: httpx.Request) -> httpx.Response:
        calls.append(str(request.url))
        return httpx.Response(302, headers={"Location": "https://hook.invalid/other"})

    result = asyncio.run(
        send_card(
            _settings(),
            _card((_line("公式", VERDICT_CORRECT),)),
            transport=_transport(handler),
            gate=_gate(),
        )
    )
    assert calls == [WEBHOOK]
    assert result.ok is False
    assert result.retryable is False


def test_client_timeout_defaults_to_ten_seconds(monkeypatch: pytest.MonkeyPatch) -> None:
    """默认超时 10 秒，不跟随重定向，不走环境变量里的代理。"""

    captured: dict[str, object] = {}
    real = httpx.AsyncClient

    def factory(*_args: object, **kwargs: object) -> httpx.AsyncClient:
        captured["timeout"] = kwargs["timeout"]
        captured["follow_redirects"] = kwargs["follow_redirects"]
        captured["trust_env"] = kwargs["trust_env"]
        kwargs["transport"] = _transport(lambda _request: httpx.Response(200, json={"code": 0}))
        return real(**kwargs)  # type: ignore[arg-type]

    monkeypatch.setattr("post_clear_check.feishu.httpx.AsyncClient", factory)
    result = asyncio.run(
        send_card(
            _settings(),
            _card((_line("公式", VERDICT_CORRECT),)),
            gate=_gate(),
        )
    )
    assert result.ok is True
    assert DEFAULT_TIMEOUT_SECONDS == 10
    assert captured["follow_redirects"] is False
    assert captured["trust_env"] is False
    assert captured["timeout"] == httpx.Timeout(10.0)

    asyncio.run(
        send_card(
            _settings(),
            _card((_line("公式", VERDICT_CORRECT),)),
            timeout_seconds=2.5,
            gate=_gate(),
        )
    )
    assert captured["timeout"] == httpx.Timeout(2.5)


def test_signed_payload_uses_stable_sign_and_hides_secret() -> None:
    """有密钥时正文带 timestamp 和 sign，报文里没有密钥和 Webhook。"""

    seen: list[dict[str, object]] = []

    def handler(request: httpx.Request) -> httpx.Response:
        seen.append(json.loads(request.content.decode("utf-8")))
        return httpx.Response(200, json={"code": 0})

    result = asyncio.run(
        send_card(
            _settings(secret="test-secret"),
            _card((_line("公式", VERDICT_ERROR),)),
            transport=_transport(handler),
            gate=_gate(),
            clock=lambda: 1599360473,
        )
    )
    assert result.ok is True
    payload = seen[0]
    assert payload["timestamp"] == "1599360473"
    assert payload["sign"] == sign_payload("1599360473", "test-secret")
    assert payload["msg_type"] == "interactive"
    encoded = json.dumps(payload, ensure_ascii=False)
    assert "test-secret" not in encoded
    assert WEBHOOK not in encoded
    assert payload["card"]["header"]["template"] == "red"


def test_unsigned_payload_has_no_timestamp() -> None:
    """没配密钥时不签名。"""

    seen: list[dict[str, object]] = []

    def handler(request: httpx.Request) -> httpx.Response:
        seen.append(json.loads(request.content.decode("utf-8")))
        return httpx.Response(200, json={"StatusCode": 0})

    result = asyncio.run(
        send_card(
            _settings(secret=""),
            _card((_line("公式", VERDICT_SUSPECT),), kind="fund_account", chain_name=CHAIN_CASH_CHANGE, biz_key="FA001"),
            transport=_transport(handler),
            gate=_gate(),
        )
    )
    assert result.ok is True
    assert "timestamp" not in seen[0]
    assert "sign" not in seen[0]
    assert seen[0]["card"]["header"]["title"]["content"] == "可疑 资金变动 资金账户 FA001"
    assert seen[0]["card"]["config"] == {"wide_screen_mode": True}


def test_empty_lines_are_not_sent() -> None:
    """没有结论行的卡片不发请求。"""

    called = False

    def handler(_request: httpx.Request) -> httpx.Response:
        nonlocal called
        called = True
        return httpx.Response(200, json={"code": 0})

    result = asyncio.run(
        send_card(
            _settings(),
            _card(()),
            transport=_transport(handler),
            gate=_gate(),
        )
    )
    assert called is False
    assert result.attempted is False
    assert result.ok is False
    assert result.retryable is False


def test_bad_url_is_not_retryable_and_log_has_no_address() -> None:
    """协议不对不是网络失败。异常原文里的地址也不能进日志。"""

    lines: list[str] = []
    handle = logger.add(lambda message: lines.append(str(message)), diagnose=False, backtrace=False)
    webhook = "ftp://hook.invalid/marker"
    try:
        result = asyncio.run(
            send_card(
                _settings(webhook=webhook, secret=FEISHU_SIGN_SECRET),
                _card((_line("公式", VERDICT_ERROR),)),
                gate=_gate(),
            )
        )
    finally:
        logger.remove(handle)
    assert result.ok is False
    assert result.retryable is False
    assert result.attempted is True
    text = "".join(lines)
    assert webhook not in text
    assert "hook.invalid" not in text
    assert FEISHU_SIGN_SECRET not in text
    assert "不可重试" in text


def test_invalid_verdict_timeout_and_kind_raise_without_request() -> None:
    """结论、超时、主键名称在发出请求前就拦住。"""

    called = False

    def handler(_request: httpx.Request) -> httpx.Response:
        nonlocal called
        called = True
        return httpx.Response(200, json={"code": 0})

    with pytest.raises(FeishuError, match="结论不在"):
        asyncio.run(
            send_card(
                _settings(),
                _card((_line("公式", "bad"),)),
                transport=_transport(handler),
                gate=_gate(),
            )
        )
    with pytest.raises(FeishuError, match="超时必须大于 0"):
        asyncio.run(
            send_card(
                _settings(),
                _card((_line("公式", VERDICT_ERROR),)),
                timeout_seconds=0,
                transport=_transport(handler),
                gate=_gate(),
            )
        )
    with pytest.raises(FeishuError, match="主键名称"):
        asyncio.run(
            send_card(
                _settings(),
                _card((_line("公式", VERDICT_ERROR),), kind="note"),
                transport=_transport(handler),
                gate=_gate(),
            )
        )
    assert called is False


def test_missing_webhook_error_has_no_secret() -> None:
    """没配 Webhook 时的异常文本不带密钥。"""

    with pytest.raises(FeishuError, match="未配置飞书 Webhook") as caught:
        asyncio.run(
            send_card(
                _settings(webhook="  ", secret=FEISHU_SIGN_SECRET),
                _card((_line("公式", VERDICT_ERROR),)),
                gate=_gate(),
            )
        )
    assert FEISHU_SIGN_SECRET not in str(caught.value)
    assert FEISHU_WEBHOOK not in str(caught.value)


def test_rate_gate_waits_instead_of_dropping() -> None:
    """同一秒第 6 次会等到窗口滑开。一分钟第 101 次同样等待，不丢。"""

    now = {"t": 0.0}

    def clock() -> float:
        return now["t"]

    async def sleep(seconds: float) -> None:
        now["t"] += seconds

    gate = RateGate(clock=clock, sleep=sleep)

    async def six() -> None:
        for _ in range(6):
            await gate.acquire()

    asyncio.run(six())
    assert now["t"] == pytest.approx(1.0)

    now["t"] = 0.0
    gate = RateGate(clock=clock, sleep=sleep)

    async def hundred_one() -> None:
        for _ in range(101):
            await gate.acquire()

    asyncio.run(hundred_one())
    assert now["t"] == pytest.approx(60.0)


def test_error_theme_posts_only_error_rows() -> None:
    """主题是错误时，正文只有错误清单。正确和可疑不出现，摘要说明它们留在核对明细。"""

    seen: list[dict[str, object]] = []

    def handler(request: httpx.Request) -> httpx.Response:
        seen.append(json.loads(request.content.decode("utf-8")))
        return httpx.Response(200, json={"code": 0})

    lines = (
        _line("正确公式", VERDICT_CORRECT, calc="2 + 2 = 4；这是正确说明", check="报表.甲 = 4"),
        _line(
            "错误公式",
            VERDICT_ERROR,
            calc="3 + 1 = 4；这是错误说明",
            check="报表.乙 = 9",
            calc_value=Decimal("4"),
            check_value=Decimal("9"),
        ),
        _line("可疑公式", VERDICT_SUSPECT, calc="字段缺失", calc_value=None, check_value=Decimal("1")),
    )
    result = asyncio.run(
        send_card(
            _settings(),
            _card(lines),
            transport=_transport(handler),
            gate=_gate(),
        )
    )
    assert result.ok is True
    assert len(seen) == 1
    card = seen[0]["card"]
    assert card["header"]["title"]["content"] == f"错误 买入执行 主订单 {ORDER_NO}"
    text = card["elements"][0]["text"]["content"]
    assert "错误公式" in text
    assert "正确公式" not in text
    assert "可疑公式" not in text
    assert "。符合预期。" not in text
    assert "无法判定。" not in text
    assert "正确与可疑留在核对明细。" in text
    assert "\n\n每单链路\n\n" not in text
    assert "\n\n盘中公式\n\n" not in text
    assert (
        "关联报表\n\n"
        "字段：错误公式\n\n"
        "期望：3 + 1 = 4\n\n"
        "实据：报表.乙 = 9。当前 9.00\n\n"
        "依据：依据\n\n"
        "结论：错误。不符合预期。这是错误说明。"
    ) in text


def test_suspect_theme_still_includes_correct_rows() -> None:
    """主题是可疑时，正文仍带上收敛后的正确行。只对错误主题收成错误清单。"""

    lines = (
        _line("正确公式", VERDICT_CORRECT),
        _line("可疑公式", VERDICT_SUSPECT, calc_value=Decimal("1"), check_value=Decimal("2")),
    )
    bodies = pack_bodies(
        title=render_title(VERDICT_SUSPECT, CHAIN_BUY, "master_order", ORDER_NO),
        verdict=VERDICT_SUSPECT,
        business_date=DAY,
        trade_date=DAY,
        batch_no="B001",
        chain_name=CHAIN_BUY,
        record_no="81",
        lines=lines,
        signed=False,
    )
    assert len(bodies) == 1
    assert "\n\n盘中公式\n\n" in bodies[0]
    assert "字段：正确公式" in bodies[0]
    assert "字段：可疑公式" in bodies[0]
    assert "结论：正确" in bodies[0]
    assert "结论：可疑。无法判定。" in bodies[0]
    assert "正确与可疑留在核对明细。" not in bodies[0]


def test_send_posts_json_shape() -> None:
    """发出去的卡片含宽屏、标题和一条 lark_md 正文。"""

    seen: list[dict[str, object]] = []

    def handler(request: httpx.Request) -> httpx.Response:
        seen.append(json.loads(request.content.decode("utf-8")))
        return httpx.Response(200, json={"code": 0})

    asyncio.run(
        send_card(
            _settings(),
            _card(
                (
                    _line(
                        "票据面值(结算币种) + 发行费用(结算币种)",
                        VERDICT_CORRECT,
                        calc="1000000.00 + 1000.00 = 1001000.00",
                        check="OTC003 共赢票据订单报表.总金额(结算币种) = 1001000.00",
                    ),
                )
            ),
            transport=_transport(handler),
            gate=_gate(),
        )
    )
    card = seen[0]["card"]
    assert card["elements"] == [
        {
            "tag": "div",
            "text": {
                "tag": "lark_md",
                "content": card["elements"][0]["text"]["content"],
            },
        }
    ]
    text = card["elements"][0]["text"]["content"]
    assert text.startswith(f"营业日：2026-10-15\n交易日：2026-10-15\n批次号：B001\n涉及：{BUY_INVOLVED}\n")
    assert "字段：票据面值(结算币种) + 发行费用(结算币种)" in text
    assert "\n\n关联报表\n\n" in text
    assert "结论：正确" in text
    assert seen[0]["card"]["header"]["title"]["content"] == f"正确 买入执行 主订单 {ORDER_NO}"
    assert len(seen) == 1


def test_same_report_is_one_section_and_errors_come_first() -> None:
    """错误清单在正确覆盖块前面。同一依据去掉列名。同行等式说明不重复。"""

    basis = "产品原型 报表.md 「列」"
    correct = Line(
        formula_text="价格 = 100",
        calc_text="100；发行端 OTC003 共赢票据订单报表同行等式",
        calc_value=Decimal("100"),
        check_text="OTC003 共赢票据订单报表.价格 = 100.00（主订单编号 ORD1）",
        check_value=Decimal("100"),
        basis=basis,
        verdict=VERDICT_CORRECT,
    )
    wrong = Line(
        formula_text="股票数量 = 名义本金(交易币种) / 成交股价",
        calc_text="1 / 2 = 0.50；回写 after_val 优先",
        calc_value=Decimal("0.50"),
        check_text="OTC003 共赢票据订单报表.股票数量 = 9.00（主订单编号 ORD1）",
        check_value=Decimal("9"),
        basis="产品原型 报表.md 「股票数量」",
        verdict=VERDICT_ERROR,
    )
    body = render_body(
        business_date=DAY,
        trade_date=DAY,
        batch_no="B001",
        chain_name=CHAIN_BUY,
        lines=(correct, wrong),
        record_no="81",
    )
    assert body.index("股票数量") < body.index("价格 = 100")
    assert "\n\n关联报表\n\n" in body
    assert body.count("依据：产品原型 报表.md") == 2
    assert "主订单编号 ORD1" in body
    assert "同行等式" not in body
    assert "回写 after_val 优先" in body
    assert "代入：100" not in body
    assert "被核对的对象" not in body


def test_chapters_keep_blocked_rows_and_not_applicable_rows() -> None:
    """期望值为空的可疑进行无法展开。公式里写了不出现的正确行进行不适用。开头四行仍在。"""

    correct = _line(
        "价格 = 100",
        VERDICT_CORRECT,
        check="OTC003 订单.价格 = 100（主订单编号 ORD1）",
        basis="产品原型 报表.md 「价格」",
    )
    other = _line(
        "数量 = 2",
        VERDICT_CORRECT,
        check="OTC003 订单.数量 = 2（主订单编号 ORD1）",
        basis="产品原型 报表.md 「数量」",
    )
    absent = _line(
        "金额为 0 的冻结必须不出现",
        VERDICT_CORRECT,
        calc="未出现",
        calc_value=None,
        check_value=None,
    )
    suspect = _line("结算汇率", VERDICT_SUSPECT, calc_value=Decimal("1.56"), check_value=Decimal("1.17"))
    blocked = _line("实时估值", VERDICT_SUSPECT, calc="最新价缺失", calc_value=None, check_value=None)
    wrong = _line("总金额", VERDICT_ERROR, calc_value=Decimal("1"), check_value=Decimal("2"))
    body = render_body(
        business_date=DAY,
        trade_date=DAY,
        batch_no="B001",
        chain_name=CHAIN_CASH_CHANGE,
        lines=(correct, other, absent, suspect, blocked, wrong),
        record_no="81",
        sample_index=2,
    )
    assert body.startswith(
        f"抽样明细第 2 条\n营业日：2026-10-15\n交易日：2026-10-15\n批次号：B001\n涉及：{CASH_INVOLVED}\n\n"
    )
    assert body.index("\n\n错误清单\n\n") < body.index("\n\n可疑清单\n\n")
    assert body.index("\n\n可疑清单\n\n") < body.index("\n\n无法展开\n\n")
    assert body.index("\n\n无法展开\n\n") < body.index("\n\n正确\n\n")
    assert body.index("\n\n正确\n\n") < body.index("\n\n不适用\n\n")
    suspect_part = body.split("\n\n可疑清单\n\n", 1)[1].split("\n\n无法展开\n\n", 1)[0]
    blocked_part = body.split("\n\n无法展开\n\n", 1)[1].split("\n\n正确\n\n", 1)[0]
    correct_part = body.split("\n\n正确\n\n", 1)[1].split("\n\n不适用\n\n", 1)[0]
    absent_part = body.split("\n\n不适用\n\n", 1)[1]
    assert "总金额" in body.split("\n\n错误清单\n\n", 1)[1].split("\n\n可疑清单\n\n", 1)[0]
    assert "结算汇率" in suspect_part
    assert "实时估值" not in suspect_part
    assert "实时估值" in blocked_part
    assert "期望值无法计算" in blocked_part
    assert "价格 = 100" in correct_part
    assert "数量 = 2" in correct_part
    assert "2 列无差异" in correct_part
    assert "筛选：主订单 ORD1" in correct_part
    assert "必须不出现" not in correct_part
    assert "必须不出现" in absent_part
    assert "结论：正确" in absent_part
    assert "本卡正确 3，错误 1，可疑 2。" in body


def test_order_chain_groups_link_formula_clear_and_report() -> None:
    """买入、卖出明细按四章排。每单链路只写结论，计算按字段列出。"""

    lines = (
        _line("执行事实列与源表一致", VERDICT_CORRECT, calc="一致", check="源表一致"),
        _line(
            "对冲记录成交股数 = TRS 执行根的成交股数",
            VERDICT_CORRECT,
            calc="600 = 600",
            check="成交股数 = 600",
        ),
        _line(
            "票据面值 = 执行订单名义本金（结算币种） × A 类报价",
            VERDICT_CORRECT,
            calc="100 × 10% = 10",
            check="面值 = 10",
        ),
        _line("T日订单结算与工作快照一致", VERDICT_CORRECT, calc="一致", check="cleared 一致"),
        _line(
            "总金额 = 票据发行价 + 承销费用",
            VERDICT_CORRECT,
            calc="10 + 1 = 11",
            check="OTC003 共赢票据订单报表.总金额 = 11",
        ),
    )
    body = render_body(
        business_date=DAY,
        trade_date=DAY,
        batch_no="B001",
        chain_name=CHAIN_BUY,
        lines=lines,
        record_no="81",
    )
    assert body.index("\n\n每单链路\n\n") < body.index("\n\n盘中公式\n\n")
    assert body.index("\n\n盘中公式\n\n") < body.index("\n\n清算对照\n\n")
    assert body.index("\n\n清算对照\n\n") < body.index("\n\n关联报表\n\n")
    assert "结论：正确。执行事实列与源表一致。对冲记录成交股数等于 TRS 执行根的成交股数。" in body
    assert "字段：执行事实列与源表一致" not in body
    assert "字段：票据面值" in body
    assert "字段：T日订单结算与工作快照一致" in body
    assert "字段：总金额" in body
    sell = render_body(
        business_date=DAY,
        trade_date=DAY,
        batch_no="B001",
        chain_name=CHAIN_SELL,
        lines=lines[2:3],
        record_no="81",
    )
    assert "\n\n盘中公式\n\n" in sell
    assert "\n\n每单链路\n\n" not in sell


def test_sample_line_is_the_first_body_line() -> None:
    """配了推送条数时，抽样序号写在营业日前面。"""

    body = render_body(
        business_date=DAY,
        trade_date=DAY,
        batch_no="B001",
        chain_name=CHAIN_BUY,
        lines=(_line("公式", VERDICT_CORRECT),),
        record_no="81",
        sample_index=1,
    )
    assert body.startswith("抽样明细第 1 条\n营业日：2026-10-15\n")


def test_body_strips_control_characters_and_angle_brackets() -> None:
    """制表符去掉，尖括号改成全角，避免 lark_md 当成标签。"""

    body = render_body(
        business_date=DAY,
        trade_date=DAY,
        batch_no="B001",
        chain_name=CHAIN_BUY,
        lines=(_line("a<b>\tc", VERDICT_ERROR, calc="1\t2", check="报<表>.甲 = 1"),),
        record_no="81",
    )
    assert "\t" not in body
    assert "<" not in body
    assert ">" not in body
    assert "a＜b＞c" in body
    assert "报＜表＞" in body


def test_failure_log_includes_feishu_msg_without_webhook() -> None:
    """业务码失败时带上飞书 msg，地址仍不落日志。"""

    lines: list[str] = []
    handle = logger.add(lambda message: lines.append(str(message)), diagnose=False, backtrace=False)

    def handler(_request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json={"code": 11246, "msg": f"invalid tag {WEBHOOK}"})

    async def run() -> None:
        result = await send_card(
            _settings(webhook=WEBHOOK),
            _card((_line("公式", VERDICT_ERROR),)),
            transport=_transport(handler),
            gate=_gate(),
        )
        assert result.ok is False
        assert result.retryable is False
        assert result.attempted is True

    try:
        asyncio.run(run())
    finally:
        logger.remove(handle)
    text = "".join(lines)
    assert "11246" in text
    assert "不可重试" in text
    assert "invalid tag" in text
    assert WEBHOOK not in text
    assert "https://" not in text
    assert "推送类型未配置，三类都推" in text


def test_oversized_errors_split_into_posts_under_byte_limit() -> None:
    """错误行全部保留，超限时拆成多张，每张请求体不超过给定字节。"""

    lines = tuple(_line(f"标记{index}" + "错" * 80, VERDICT_ERROR) for index in range(4))
    title = render_title(VERDICT_ERROR, CHAIN_BUY, "master_order", ORDER_NO)
    limit = 2200
    bodies = pack_bodies(
        title=title,
        verdict=VERDICT_ERROR,
        business_date=DAY,
        trade_date=DAY,
        batch_no="B001",
        chain_name=CHAIN_BUY,
        record_no="81",
        lines=lines,
        signed=True,
        limit=limit,
    )
    assert len(bodies) >= 2
    seen: set[str] = set()
    for index, body in enumerate(bodies, start=1):
        assert f"第{index}张，共{len(bodies)}张" in body
        assert "记录编号：81" in body
        # 拆分以段为单位，每张都保留开头四行（含「涉及」）。
        assert f"\n涉及：{BUY_INVOLVED}\n" in body
        payload = build_payload(
            title=title,
            body=body,
            verdict=VERDICT_ERROR,
            timestamp="1700000000",
            sign="A" * 44,
        )
        assert len(json.dumps(payload).encode("utf-8")) <= limit
        for marker in range(4):
            token = f"标记{marker}"
            if token in body:
                seen.add(token)
    assert seen == {f"标记{index}" for index in range(4)}

    posted: list[bytes] = []

    def handler(request: httpx.Request) -> httpx.Response:
        posted.append(request.content)
        return httpx.Response(200, json={"code": 0})

    result = asyncio.run(
        send_card(
            _settings(),
            _card(tuple(_line(f"长错误{index}" + "错" * 800, VERDICT_ERROR) for index in range(6))),
            transport=_transport(handler),
            gate=_gate(),
        )
    )
    assert result.ok is True
    assert len(posted) >= 2
    for raw in posted:
        assert len(raw) <= 28 * 1024
    joined = "".join(item.decode("utf-8") for item in posted)
    for index in range(6):
        assert f"长错误{index}" in joined


def _owner(
    kind: str,
    biz_key: str,
    verdict: str,
    lines: tuple[Line, ...] = (),
) -> SummaryOwner:
    return SummaryOwner(kind, biz_key, verdict, lines)


def test_summary_lists_owners_by_verdict_and_kind() -> None:
    """概况按正确、错误、可疑三行写归属。主键按字典序，归属顺序固定，颜色取最严重的一档。"""

    owners = (
        _owner("fund_account", "FA2", VERDICT_ERROR),
        _owner("master_order", "ORD2", VERDICT_CORRECT),
        _owner("fund_account", "FA9", VERDICT_CORRECT),
        _owner("master_order", "ORD1", VERDICT_CORRECT),
        _owner("trs_user", "U9", VERDICT_SUSPECT),
        _owner("fund_account", "FA1", VERDICT_ERROR),
    )
    assert render_summary_title(CHAIN_TRADE, owners) == "链路-委托执行 正确3条，错误2条，可疑1条"
    assert render_summary_title(CHAIN_TRADE, owners, resend=True).endswith(" 重发")
    assert summary_verdict(owners) == VERDICT_ERROR
    assert render_summary_body(owners) == (
        "正确：主订单：ORD1、ORD2；资金账户：FA9\n"
        "\n"
        "错误：资金账户：FA1、FA2\n"
        "\n"
        "可疑：用户账号：U9"
    )


def test_summary_writes_none_when_a_verdict_is_absent() -> None:
    """没有的那一档写「无」，三行都在。没有结论行时不编示例。"""

    owners = (_owner("master_order", "ORD1", VERDICT_CORRECT),)
    body = render_summary_body(owners)
    assert body == "正确：主订单：ORD1\n\n错误：无\n\n可疑：无"
    assert "示例解读" not in body
    assert render_summary_title(CHAIN_BUY, owners) == "链路-买入执行 正确1条，错误0条，可疑0条"
    assert summary_verdict(owners) == VERDICT_CORRECT


def test_summary_ignores_detail_lines() -> None:
    """归属上带了结论行也不写进概况。三档只列主键。"""

    line = _line(
        "总金额 = 1",
        VERDICT_CORRECT,
        calc="1.00 + 0.00 = 1.00；四舍五入",
        check="总金额 = 1.00",
    )
    body = render_summary_body((_owner("master_order", "ORD1", VERDICT_CORRECT, (line,)),))
    assert body == "正确：主订单：ORD1\n\n错误：无\n\n可疑：无"
    assert "示例解读" not in body
    assert "总金额" not in body
    assert "依据" not in body


def test_summary_escapes_angle_brackets_in_keys() -> None:
    """主键里的尖括号改成全角，避免 lark_md 当成标签。"""

    body = render_summary_body(
        (_owner("master_order", "<ORD>", VERDICT_ERROR, (_line("错公式", VERDICT_ERROR),)),)
    )
    assert body == "正确：无\n\n错误：主订单：＜ORD＞\n\n可疑：无"
    assert "<ORD>" not in body


def test_summary_truncates_correct_names_before_errors() -> None:
    """超长时先截正确名单，错误主键和「可疑：无」都留着。标题条数仍是原长。"""

    owners = tuple(
        _owner("master_order", f"K{index:020d}", VERDICT_CORRECT) for index in range(200)
    ) + (_owner("master_order", "ERRKEY", VERDICT_ERROR),)
    body = render_summary_body(owners)
    correct, error, suspect = body.split("\n\n")
    assert len(body) <= BODY_LIMIT
    assert correct.startswith("正确：主订单：")
    assert "…等共200个" in correct
    assert error == "错误：主订单：ERRKEY"
    assert suspect == "可疑：无"
    assert render_summary_title(CHAIN_BUY, owners) == "链路-买入执行 正确200条，错误1条，可疑0条"


def test_summary_post_uses_chain_title_and_owner_body() -> None:
    """概况请求是一张 lark_md。有错误时标题是红的，正文只列归属，不写营业日和示例。"""

    seen: list[dict[str, object]] = []

    def handler(request: httpx.Request) -> httpx.Response:
        seen.append(json.loads(request.content.decode("utf-8")))
        return httpx.Response(200, json={"code": 0})

    card = OutboundCard(
        card_kind=CARD_SUMMARY,
        chain_name=CHAIN_BUY,
        biz_key=CHAIN_BUY,
        business_date=DAY,
        trade_date=DAY,
        batch_no="B001",
        lines=(),
        record_no="9",
        owners=(
            _owner("master_order", "ORD2", VERDICT_CORRECT, (_line("对公式", VERDICT_CORRECT),)),
            _owner("master_order", "ORD1", VERDICT_ERROR, (_line("错公式", VERDICT_ERROR),)),
        ),
    )
    result = asyncio.run(send_summary(_settings(), card, transport=_transport(handler), gate=_gate()))
    assert result.ok is True
    assert len(seen) == 1
    header = seen[0]["card"]["header"]
    assert header["template"] == "red"
    assert header["title"]["content"] == "链路-买入执行 正确1条，错误1条，可疑0条"
    text = seen[0]["card"]["elements"][0]["text"]["content"]
    assert text == "正确：主订单：ORD2\n\n错误：主订单：ORD1\n\n可疑：无"
    assert "对公式" not in text
    assert "错公式" not in text
    assert "营业日" not in text


def test_summary_without_owners_is_not_posted() -> None:
    """没有归属就不发请求。"""

    def handler(_request: httpx.Request) -> httpx.Response:
        raise AssertionError("不应该发出请求")

    card = OutboundCard(
        card_kind=CARD_SUMMARY,
        chain_name=CHAIN_BUY,
        biz_key=CHAIN_BUY,
        business_date=DAY,
        trade_date=DAY,
        batch_no="B001",
        lines=(),
        record_no="9",
    )
    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_link_summary_describes_facts_and_hedge_roots() -> None:
    """每单链路写成执行事实、对冲执行根两段，不把结论收成正确几笔。"""

    body = render_body(
        business_date=DAY,
        trade_date=DAY,
        batch_no="IB",
        chain_name=DIM_ORDER,
        record_no="13",
        lines=(
            _dim("执行事实列与源表一致", VERDICT_CORRECT, master_order_no="ORD1", calc="一致", calc_value=None, check_value=None),
            _dim("执行事实列与源表一致", VERDICT_CORRECT, master_order_no="ORD2", calc="一致", calc_value=None, check_value=None),
            _dim(
                "对冲记录成交股数 = TRS 执行根的成交股数",
                VERDICT_CORRECT,
                master_order_no="ORD1",
                calc="db_trs.trade_order.ext_ref=HR66 600",
                calc_value=Decimal("600"),
                check_value=Decimal("600"),
            ),
            _dim(
                "对冲记录成交股数 = TRS 执行根的成交股数",
                VERDICT_ERROR,
                master_order_no="ORD2",
                calc="db_trs.trade_order.ext_ref=HR70 没有行",
                calc_value=None,
                check_value=Decimal("100"),
            ),
        ),
    )
    assert "已完成或已结算的 ORD 主单 2 笔。执行事实核对 input 封存的成交名义本金、订单成交均价与实时源表是否一致：2 笔一致。" in body
    assert "需要 TRS 执行根的对冲 2 条，成交股数与执行根对照：1 条一致。" in body
    assert "自动单按 ext_ref（HR 加对冲主键）定位，补录按委托号或算法母单号。" in body
    assert "自定义补录和从未建单不要求执行根。" in body
    assert "ORD2：db_trs.trade_order.ext_ref=HR70 没有行" in body
    assert "正确 8 笔，错误 0 笔，可疑 0 笔" not in body


def test_order_card_expands_clearing_and_related_reports() -> None:
    """清算对照和关联报表按巡检稿的字段展开，不把多列收成一行笔数。"""

    body = render_body(
        business_date=DAY,
        trade_date=DAY,
        batch_no="IB",
        chain_name=DIM_ORDER,
        record_no="14",
        lines=(
            _dim(
                "T日订单结算与工作快照一致",
                VERDICT_CORRECT,
                master_order_no="ORD1",
                calc="一致",
                calc_value=None,
                check_value=None,
            ),
            _dim(
                f"{CLEAR_MARK}交收台账",
                VERDICT_CORRECT,
                calc="字段：交收台账\n期望：交收推演\n实据：交收台账：本轮主单 8 笔，正确 8，错误 0，可疑 0。\n依据：02 方案交收\n结论：正确（8/8）",
                calc_value=None,
                check_value=None,
            ),
            _dim(
                "关联报表索引：发行 OTC003",
                VERDICT_CORRECT,
                calc="字段：发行 OTC003\n期望：名义本金 = 执行名义\n实据：发行 OTC003：本轮主单 8 笔，正确 8，错误 0，可疑 0。\n依据：产品原型 OTC003\n结论：正确（8/8）",
                calc_value=None,
                check_value=None,
            ),
        ),
    )
    assert "清算对照\n字段：T日订单结算与工作快照" in body
    assert "字段：交收台账\n期望：交收推演" in body
    assert "关联报表\n字段：发行 OTC003\n期望：名义本金 = 执行名义" 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_report_bases_are_one_per_line() -> None:
    """多条依据各占一行。分号拼在一起时，原型路径会挤成一行。"""

    body = render_body(
        business_date=DAY,
        trade_date=DAY,
        batch_no="IB",
        chain_name=DIM_SALES,
        record_no="12",
        lines=(
            _dim(
                "股票数量",
                VERDICT_CORRECT,
                check="OTC003 共赢票据订单报表.股票数量 = 1",
                report_code="OTC003",
                basis="产品原型 订单报表.md 「股票数量」",
            ),
            _dim(
                "价格",
                VERDICT_CORRECT,
                check="OTC003 共赢票据订单报表.价格 = 1",
                report_code="OTC003",
                basis="产品原型 订单报表.md 「价格」",
            ),
            _dim(
                "总金额",
                VERDICT_CORRECT,
                check="OTC003 共赢票据订单报表.总金额 = 1",
                report_code="OTC003",
                basis="产品原型 订单报表.md 「总金额」",
            ),
        ),
    )
    assert "依据：产品原型 订单报表.md 「股票数量」\n产品原型 订单报表.md 「价格」\n产品原型 订单报表.md 「总金额」" in body
    assert "另有" not in body
    assert "依据：产品原型 订单报表.md 「股票数量」；" not in body
    assert "期望：股票数量\n价格\n总金额" 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