inquiry_robot/inquiry-agent/agent/policy/inquiry_copy.py

"""
空运文字询价出站话术（销售可见）。

本文件职责：集中文案；按钮字面量必须与 handler 精确匹配。
禁止：在本文件调主账/TMS；禁止把密钥写进话术。
"""

from __future__ import annotations

import json
import re
import time
import uuid
from typing import Any, Iterable

from agent.schema.field_validate import (
    AIR_COLLAB_FIELDS,
    AIR_OPTIONAL_FIELDS,
    AIR_REQUIRED_FIELDS,
    LAND_DISPLAY,
    SEA_COLLAB_FIELDS,
    SEA_REQUIRED_FIELDS,
    collab_display_name,
    collab_fields_for_ticket,
    display_name,
)
from agent.schema.land_options import option_prompt as _land_option_prompt
from agent.schema.quote_adjust import total_price_label

# 空运补问：必填 7 项 + 非必填。报价日期默认当天，不进待补充。
_AIR_EXAMPLES = {
    "起运港": "南京",
    "目的港": "吉隆坡",
    "品名": "茶叶",
    "件数": "10",
    "毛重": "100",
    "体积": "2",
    "包装方式": "托盘、散货",
    "报价日期": "2026-09-15",
    "货源地": "东莞",
    "货好时间": "随时可提",
    "提货地址": "东莞仓库",
    "收货地址": "马尼拉",
    "贸易条款": "DDP、DAP、CIF、CIP",
}
_AIR_CLARIFY = tuple(
    (key, label, _AIR_EXAMPLES[key]) for key, label in AIR_REQUIRED_FIELDS
)
_AIR_OPTIONAL_CLARIFY = tuple(
    (key, label, _AIR_EXAMPLES.get(key, "")) for key, label in AIR_OPTIONAL_FIELDS
)
AIR_CONFIRM_TAIL = "请回复“确定”进行TMS查价，内容有误时，请重新发内容。"
AIR_CONFIRM_EXPIRED = "刚才的空运核对已过期，请把空运需求再发一次，出核对卡后再回复确定。"

# 海运补问：与合同 10 项一致。报价日期默认当天，不进待补充。
_SEA_EXAMPLES = {
    "起运港": "上海",
    "目的港": "洛杉矶",
    "品名": "普货",
    "货量": "20吨",
    "整柜或拼柜": "整柜",
    "箱型箱量": "1x40HQ",
    # 销售看到的举例是门到门、港到港，不是 FOB。举例行不参与已填判断。
    "贸易条款": "门到门、港到港",
    "运输分类": "Port to Port",
    "报价日期": "2026-09-15",
}
_SEA_CLARIFY = tuple(
    (key, label, _SEA_EXAMPLES[key]) for key, label in SEA_REQUIRED_FIELDS
)

# 文字按钮（销售点卡或原样回复均可）
BTN_SKIP_COLLAB = "不需要协同，直接生成报价单"
BTN_SKIP_COLLAB_OLD = "不需要协同，直接生成基础报价单"
# 海运/陆运已有 TMS 报价时价格明细卡用短文案；空运仍用上面长文案。点卡 EventKey 不变。
BTN_SKIP_COLLAB_TMS = "直接生成报价单"
BTN_SKIP_COLLAB_KEY = "skip_collab"
BTN_SKIP_COLLAB_DONE = "已提交"
BTN_PULL_COLLAB = "拉产品进群协同"
BTN_PULL_COLLAB_KEY = "pull_collab"
BTN_ADOPT_EXCEL = "采用，生成Excel报价单"
BTN_ADOPT_PDF = "采用，生成PDF报价单"
BTN_REJECT = "不采用"
# 企微一张卡三个按钮只能横排，长文案会被裁。卡片用短文案才能看全。
BTN_ADOPT_EXCEL_CARD = "生成Excel"
BTN_ADOPT_PDF_CARD = "生成PDF"
BTN_ADOPT_EXCEL_KEY = "adopt_excel"
BTN_ADOPT_PDF_KEY = "adopt_pdf"
BTN_REJECT_KEY = "reject"
BTN_DEAL = "成交"
BTN_LOST = "未成交"
BTN_NEGOTIATE = "协商中"
BTN_DEAL_KEY = "deal"
BTN_LOST_KEY = "lost"
BTN_NEGOTIATE_KEY = "negotiate"
# 旧卡 EventKey 兼容；新暂无报价卡按现网图：无按钮。
BTN_COPY_TICKET = "复制工单号"
BTN_COPY_TICKET_KEY = "copy_ticket"

_WO_NO = re.compile(r"(?:WO|VT)\d{12}", re.IGNORECASE)
# 仅给无网 stub / 工单号回填用，正式续办意图由 DeepSeek A 判断，不要求销售说这些词。
_CONTINUE_HINT = (
    "继续",
    "接着",
    "回到",
    "再看",
    "再出",
    "找回",
    "重开",
    "那单",
    "这单",
    "历史工单",
    "咨询工单",
    "刚才那",
    "上次那",
)


ASK_TRANSPORT = (
    "请回复数字或文字，告知我询价的运输方式：\n"
    "1.陆运\n"
    "2.海运\n"
    "3.空运"
)
# 待补充清单：请销售整段复制，只在冒号后填值；示例行不算已填。
CLARIFY_PASTE_HINT = "请复制粘贴以下内容，通过回复数字或文字方式，告知我以下字段值："
_PENDING_UNITS = {}
HANDOFF_ALREADY_COLLAB = "该工单已拉产品进群协同。"
HANDOFF_CLARIFY_REASON = "参数提取不全（连续 3 次反问仍不完整）"
HANDOFF_INTENT_REASON = "已识别您希望转人工处理"
CLARIFY_LIMIT_HINT = "连续 3 次反问仍不完整将触发转人工"
# 私聊图片询价：先回识别中；看不清/失败与补问拼成一条，销售侧不加 [test]。
IMAGE_RECOGNIZING = "正在识别图片"
IMAGE_UNREADABLE = "图片没看清询价信息，请换一张更清楚的，或直接打字。"
IMAGE_TECH_FAIL = "图片暂时没认出来，请再发一张或直接打字。"
# 私聊附件询价：H5 或发图开始认；看不清/失败与补问拼成一条。
ATTACH_RECOGNIZING = "正在识别附件"
ATTACH_UNREADABLE = "附件没看清询价信息，请换一份更清楚的 Excel / PDF（或把图片发到会话），或直接打字。"
ATTACH_TECH_FAIL = "附件暂时没认出来，请再传一次或直接打字。"
ATTACH_H5_REJECT_TYPE = "只接受 Excel、PDF；图片请发到当前企微会话。"
ATTACH_H5_NEED_FILE = "请选择要上传的 Excel 或 PDF。"
ATTACH_H5_EXPIRED = "已过期，请回企微重新说附件询价。"
ATTACH_H5_OK = "已收到，请回企微查看识别结果。"


def attachment_upload_card(url: str) -> dict[str, Any]:
    """
    附件上传入口模板卡（text_notice 跳转 H5）。

    必须跟现网同一卡型：企微 news_notice 强制要 card_image.url，缺了会 42044，
    出站再回退成一句「上传询价附件」，销售侧就看不到卡片。
    文案对齐规格：Excel/PDF 走 H5，图片发当前会话。url 必须已含 token、不含 userid。
    """
    jump = (url or "").strip()
    fallback = "https://ai.ytd-scm.com/inquiry-form"
    action_url = jump or fallback
    card: dict[str, Any] = {
        "card_type": "text_notice",
        "source": {"desc": "附件询价", "desc_color": 0},
        "main_title": {
            "title": "上传询价附件",
            "desc": "附件入口支持 Excel、PDF；图片请直接在企微聊天窗口发送",
        },
        "sub_title_text": (
            "Excel、PDF 请点击下方入口上传；图片请直接在当前企微会话发送。"
            "字段完整时自动进入建单与查价；缺失字段会在企微中补问"
        ),
        "jump_list": [{"type": 1, "title": "上传询价附件", "url": action_url}],
        "card_action": {"type": 1, "url": action_url},
    }
    return {"msgtype": "template_card", "template_card": card}


def image_fail_then_clarify(reason: str, clarify: str) -> str:
    """
    看不清或技术失败时，说明 + 文字询价补问写在同一条回复。

    禁止只回说明就把人丢在那里。两端空白去掉，中间空一行。
    """
    head = (reason or "").strip()
    body = (clarify or "").strip()
    if head and body:
        return f"{head}\n\n{body}"
    return head or body

# 旧一句只问运输类型；销售要一次看到类型、线路、分类。残留调用也走完整清单。
ASK_LAND_SUBTYPE = _land_option_prompt(current={})


def land_sample_image_payload() -> dict[str, Any]:
    """
    填写样例表图片载荷。编号清单上线后不再主动出站，保留给旧调用。

    走企微 image，不走文件气泡。禁止在回调线程上传。
    """
    from agent.schema.land_options import sample_image_path

    path = sample_image_path()
    return {
        "msgtype": "image",
        "filename": "填写样例.png",
        "filepath": str(path),
    }


def land_sample_image_fallback() -> str:
    """旧图失败兜底。新补问不再发图，一般不会走到这里。"""
    from agent.schema.land_options import sample_fallback_text

    return sample_fallback_text()
FIRST_CHAT = "你好，我是询价机器人。请直接描述客户需求，例如「上海到洛杉矶空运，3 件 180kg，下周三前到」，我会解析需求并查询 TMS 标准报价。"
ROUTE_MODEL_FAILED = "刚才没听清，请再说一次：是新询价，还是要接着办哪张工单？"
LAND_CONFIRM_TAIL = "请回复“确定”进行TMS查价，内容有误时，请重新发内容。"
LAND_CONFIRM_EXPIRED = "刚才的陆运核对已过期，请把陆运需求再发一次，出核对卡后再回复确定。"
LAND_TMS_RETRY = "工单 {wo} 上次查价没成功。回复「确定」按原需求再查一次，不用重新填。"
LAND_TMS_RETRYING = "正在重新查询TMS报价。工单：{wo}"

_LAND_LCL_KEYS = ("起运港", "目的港", "品名", "体积", "毛重", "报价日期")
_LAND_FTL_KEYS = ("起运港", "目的港", "品名", "车型/数量", "报价日期")
_LAND_HK_FULL_KEYS = ("起运港", "目的港", "品名", "报价日期")
_LAND_FTL_TYPES = frozenset({"国内运输整车", "跨境整车", "中港零担/集拼"})
_LAND_CUSTOMS_TYPES = frozenset({"跨境集拼", "跨境整车", "中港整车", "中港零担/集拼"})
_LAND_EXAMPLES = {
    "起运港": "广州",
    "目的港": "深圳",
    "品名": "衣服",
    "体积": "100CBM",
    "毛重": "50kg",
    "车型/数量": "9米6/1",
    "通关口岸": "皇岗",
    "运输类型": "国内运输拼车",
    "线路类别": "国内长途/零担",
    "运输分类": "急件",
}


def _land_cargo_keys(land_type: str, facts: dict[str, str] | None = None) -> tuple[str, ...]:
    """当前运输类型要展示的货物内部键。中港整车不要车型/数量。"""
    _ = facts
    if land_type == "中港整车":
        return _LAND_HK_FULL_KEYS
    if land_type in _LAND_FTL_TYPES:
        return _LAND_FTL_KEYS
    return _LAND_LCL_KEYS


def land_confirm_card(facts: dict[str, str] | None) -> str:
    """
    确定查价前的核对卡。没有工单号。

    行列出运输类型/线路/分类和该类型货物字段。
    国内不出现通关口岸；跨境/中港通关口岸非必填，没填留空行。
    不再单独列出中港装载方式。
    """
    src = dict(facts or {})
    land_type = (src.get("运输类型") or "").strip()
    lines = [
        f"运输类型（必填）：{land_type}",
        f"线路类别（必填）：{(src.get('线路类别') or '').strip()}",
        f"运输分类（必填）：{(src.get('运输分类') or '').strip()}",
    ]
    for key in _land_cargo_keys(land_type, src):
        label = LAND_DISPLAY.get(key, key)
        val = (src.get(key) or "").strip()
        lines.append(f"{label}（必填）：{val}")
    if land_type in _LAND_CUSTOMS_TYPES:
        lines.append(f"通关口岸（非必填）：{(src.get('通关口岸') or '').strip()}")
    lines.append("")
    lines.append(LAND_CONFIRM_TAIL)
    return "\n".join(lines)


def land_group_no_staff() -> str:
    """没有对上线路的陆运产品，不建空群。"""
    return "没有可拉的产品人员。请先在后台「员工管理」维护本线路的陆运产品岗后再试。"


def land_group_pair_rejected() -> str:
    """类型+线路不在填写样例里，禁止拉群。"""
    return "拉群失败，请排查填写内容，并重新发起工单。"


def _land_optional_keys(land_type: str) -> tuple[str, ...]:
    """当前类型询价非必填内部键。国内没有通关口岸。"""
    if land_type in _LAND_CUSTOMS_TYPES:
        return ("通关口岸",)
    return ()


def _pending_field_block(
    index: int,
    label: str,
    example: str = "",
    *,
    optional: bool = False,
    unit: str = "",
) -> str:
    """
    待补充一项：编号 + 字段名 + 必填/单位，示例另起一行。

    销售侧如：
    1.件数（必填）：
    (如：10）
    2.重量（KG）（必填）：
    (如：100）
    复制回去时只认冒号后面的值；下一行「(如：…）」不算已填。
    """
    mark = "（非必填）" if optional else "（必填）"
    extra = f"（{unit}）" if (unit or "").strip() else ""
    head = f"{index}.{label}{extra}{mark}："
    shown = (example or "").strip()
    if shown:
        return f"{head}\n(如：{shown}）"
    return head


def _land_field_line(key: str, value: str, *, required: bool, example: str = "") -> str:
    """已识别一行：名称（必填/非必填）：值。待补充走 _pending_field_block。"""
    label = LAND_DISPLAY.get(key, key)
    mark = "必填" if required else "非必填"
    if value:
        return f"{label}（{mark}）：{value}"
    _ = example
    return f"{label}（{mark}）："


def clarify_progress_line(retry: int) -> str:
    """补问次数提醒：参数提取不全（1/3）。"""
    n = int(retry or 0)
    if n < 1:
        n = 1
    if n > 3:
        n = 3
    return f"参数提取不全（{n}/3）。{CLARIFY_LIMIT_HINT}"


def with_clarify_progress(body: str, retry: int) -> str:
    """把次数提醒顶在补问正文前面。第一次补问（retry=0）不打（1/3）。"""
    text = (body or "").strip()
    n = int(retry or 0)
    if n < 1:
        return text
    head = clarify_progress_line(n)
    if not text:
        return head
    return head + "\n" + text


def ask_land_fields(
    *,
    facts: dict[str, str] | None = None,
    missing_keys: Iterable[str] = (),
    land_subtype: str = "",
) -> str:
    """
    陆运货物缺项补问。不出核对卡、不出现确定句。

    已识别和待补充都标明必填/非必填。跨境/中港空着的通关口岸也列进待补充。
    报价日期不进待补充。
    """
    src = dict(facts or {})
    skip = {"报价日期"}
    missing = [k for k in missing_keys if k and k not in skip]
    land_type = (land_subtype or src.get("运输类型") or "").strip()
    optional = _land_optional_keys(land_type)
    lines = ["请确认并补充以下信息：", "运输方式：陆运"]
    if land_type:
        lines.append(f"运输类型：{land_type}")
    for key in ("线路类别", "运输分类"):
        val = (src.get(key) or "").strip()
        if val:
            lines.append(_land_field_line(key, val, required=True))
    for key in _land_cargo_keys(land_type, src):
        val = (src.get(key) or "").strip()
        if val:
            lines.append(_land_field_line(key, val, required=True))
    for key in optional:
        val = (src.get(key) or "").strip()
        if val:
            lines.append(_land_field_line(key, val, required=False))
    pending = list(missing)
    for key in optional:
        if not (src.get(key) or "").strip() and key not in pending:
            pending.append(key)
    if not pending:
        return ""
    lines.append("")
    lines.append(CLARIFY_PASTE_HINT)
    seq = 0
    for key in pending:
        seq += 1
        required = key not in optional
        example = _LAND_EXAMPLES.get(key, "")
        label = LAND_DISPLAY.get(key, key)
        lines.append(
            _pending_field_block(
                seq,
                label,
                example,
                optional=not required,
                unit=_PENDING_UNITS.get(key, ""),
            )
        )
    return "\n".join(lines)


def _air_known_value(facts: dict[str, str], key: str) -> str:
    """已识别字段的销售侧展示值；没有则空串。"""
    src = facts or {}
    if key == "起运港":
        return (src.get("起运港") or "").strip()
    if key == "目的港":
        return (src.get("目的港") or "").strip()
    if key == "品名":
        return (src.get("品名") or "").strip()
    if key == "件数":
        raw = (src.get("件数") or "").strip()
        return _keep_or_add_unit(raw, ("件", "PCS"), "件") if raw else ""
    if key == "毛重":
        raw = (src.get("毛重") or "").strip()
        return _keep_or_add_unit(raw, ("KGS", "KG", "公斤", "千克"), "") if raw else ""
    if key == "体积":
        raw = (src.get("体积") or "").strip()
        return _keep_or_add_unit(raw, ("CBM", "M3", "M³", "立方米", "立方", "方"), "") if raw else ""
    if key == "包装方式":
        return (src.get("包装方式") or src.get("包装类型") or "").strip()
    if key == "报价日期":
        return (src.get("报价日期") or "").strip()
    from agent.llm.extract_text import clip_inquiry_value

    return clip_inquiry_value(src.get(key) or "")


def _sea_known_value(facts: dict[str, str], key: str) -> str:
    """海运已识别字段的销售侧展示值。"""
    src = facts or {}
    if key == "货量":
        return (src.get("货量") or src.get("货物数量") or "").strip()
    return (src.get(key) or "").strip()


def _air_field_line(key: str, value: str, *, required: bool, example: str = "") -> str:
    """空运核对卡一行：名称（必填/非必填）：值。待补充走 _pending_field_block。"""
    label = AIR_DISPLAY_LABEL(key)
    mark = "必填" if required else "非必填"
    text = (value or "").strip()
    if text:
        return f"{label}（{mark}）：{text}"
    _ = example
    return f"{label}（{mark}）："


def AIR_DISPLAY_LABEL(key: str) -> str:
    """核对/补问对外名。起运港按产品口径写起运港。"""
    if key == "起运港":
        return "起运港"
    return display_name(key, "AIR")


def air_confirm_card(facts: dict[str, str] | None) -> str:
    """
    确定查价前的核对卡。没有工单号。

    必填 7 项 + 6 个非必填都列出；空着的非必填留空行。
    """
    src = dict(facts or {})
    lines = ["运输方式（必填）：空运"]
    for key, _label in AIR_REQUIRED_FIELDS:
        if key == "报价日期":
            val = _air_known_value(src, key) or (src.get("报价日期") or "").strip()
            lines.append(_air_field_line(key, val, required=True))
            continue
        lines.append(_air_field_line(key, _air_known_value(src, key), required=True))
    for key, _label in AIR_OPTIONAL_FIELDS:
        lines.append(_air_field_line(key, _air_known_value(src, key), required=False))
    lines.append("")
    lines.append(AIR_CONFIRM_TAIL)
    return "\n".join(lines)


def ask_clarify(
    *,
    facts: dict[str, str] | None = None,
    missing_keys: Iterable[str] = (),
    transport_mode: str = "空运",
) -> str:
    """
    缺项补问：先回显已识别，再给可复制清单；示例另起一行，如「1.件数（必填）：」下一行「(如：10）」。

    报价日期写进已识别（销售说了用原话，没说用当天），不出现在待补充。
    空运非必填一并列出，没填进待补充，不挡查价。
    """
    src = dict(facts or {})
    skip = {"报价日期"}
    missing = [k for k in missing_keys if k and k not in skip]
    missing_set = set(missing)
    sea = (transport_mode or "").strip() == "海运"
    rows = _SEA_CLARIFY if sea else _AIR_CLARIFY
    known = _sea_known_value if sea else _air_known_value
    lines = ["请确认并补充以下信息："]
    if transport_mode:
        lines.append(f"运输方式：{transport_mode}")
    for key, label, _example in rows:
        val = known(src, key)
        if val:
            lines.append(f"{label}：{val}")
    if not sea:
        origin = known(src, "起运港")
        if origin:
            lines.append(f"起运地：{origin}")
        for key, label, _example in _AIR_OPTIONAL_CLARIFY:
            if key == "起运港":
                continue
            val = known(src, key)
            if val:
                lines.append(f"{label}：{val}")
    lines.append("")
    lines.append(CLARIFY_PASTE_HINT)
    any_missing = False
    seq = 0
    if "运输方式" in missing_set:
        any_missing = True
        seq += 1
        lines.append(_pending_field_block(seq, "运输方式", "空运"))
    for key, label, example in rows:
        if key in missing_set:
            any_missing = True
            seq += 1
            if sea and key == "运输分类":
                lines.append(_pending_field_block(seq, "运输分类", "Port to Port、其他"))
            else:
                lines.append(
                    _pending_field_block(
                        seq,
                        label,
                        example,
                        unit=_PENDING_UNITS.get(key, ""),
                    )
                )
    if not sea:
        optional_pending = [
            (key, label, example)
            for key, label, example in _AIR_OPTIONAL_CLARIFY
            if not known(src, key)
        ]
        for key, label, example in optional_pending:
            any_missing = True
            seq += 1
            shown = "起运港" if key == "起运港" else label
            lines.append(_pending_field_block(seq, shown, example, optional=True))
    if not any_missing:
        return ""
    return "\n".join(lines)


def ask_missing(display_names: Iterable[str]) -> str:
    """兼容旧调用：仅有展示名时仍走补问，但没有已识别回显。"""
    alias = {
        "起运地": "起运港",
        "目的地": "目的港",
        "起运地（机场名或三字码）": "起运港",
        "目的地（机场名或三字码）": "目的港",
        "毛重": "毛重",
        "重量": "毛重",
        "重量(KG)": "毛重",
        "重量（KG）": "毛重",
        "体积": "体积",
        "体积(CBM)": "体积",
        "体积（CBM）": "体积",
        "包装类型": "包装方式",
        "包装方式": "包装方式",
        "货物数量": "货量",
        "货量": "货量",
        "整柜或拼柜": "整柜或拼柜",
        "箱型箱量": "箱型箱量",
    }
    keys: list[str] = []
    for name in display_names:
        key = alias.get(name, name)
        if key and key not in keys and key != "报价日期":
            keys.append(key)
    return ask_clarify(facts={}, missing_keys=keys, transport_mode="空运")


def _keep_or_add_unit(raw: str, known: tuple[str, ...], default: str) -> str:
    """
    销售侧单位原样回显。

    写了 1.5KG 就回 1.5KG，禁止改成 1.5KGS。没写单位才补 default（可空）。
    """
    text = (raw or "").strip()
    if not text:
        return "-"
    upper = text.upper()
    for unit in known:
        if upper.endswith(unit.upper()):
            return text
    if default and not any(ch.isalpha() or ch in "立方件" for ch in text):
        return f"{text}{default}"
    return text


def _with_unit(raw: str, suffix: str) -> str:
    """兼容旧调用：已带单位则原样。重量不要走这里补 KGS。"""
    if suffix.upper() == "KGS":
        return _keep_or_add_unit(raw, ("KGS", "KG", "公斤", "千克"), "")
    if suffix.upper() == "CBM":
        return _keep_or_add_unit(raw, ("CBM", "M3", "M³", "立方米", "立方", "方"), "")
    return _keep_or_add_unit(raw, (suffix,), suffix)


def inquiry_field_lines(
    *,
    work_order_no: str,
    facts: dict[str, str],
    transport_mode: str = "空运",
) -> list[str]:
    """
    询价确认字段行（销售可见）。

    起运/目的展示销售原文：说南京就南京，图上/原话是 SZX 就展示 SZX。
    originCode 只给出站 TMS，不拿来改写成城市名再给销售看。
    """
    origin = (facts.get("起运港") or facts.get("起运地") or "-").strip()
    dest = (facts.get("目的港") or facts.get("目的地") or "-").strip()
    cargo = (facts.get("品名") or "-").strip() or "-"
    pack = (facts.get("包装方式") or facts.get("包装类型") or "-").strip() or "-"
    quote_date = (facts.get("报价日期") or "-").strip() or "-"
    if (transport_mode or "").strip() == "海运":
        qty = (facts.get("货量") or facts.get("货物数量") or "-").strip() or "-"
        load = (facts.get("整柜或拼柜") or "-").strip() or "-"
        box = (facts.get("箱型箱量") or "-").strip() or "-"
        trade = (facts.get("贸易条款") or "-").strip() or "-"
        klass = (facts.get("运输分类") or "-").strip() or "-"
        return [
            f"工单号：{work_order_no}",
            f"运输方式：{transport_mode}",
            f"起运港：{origin}",
            f"目的港：{dest}",
            f"品名：{cargo}",
            f"货物数量：{qty}",
            f"整柜或拼柜：{load}",
            f"箱型箱量：{box}",
            f"贸易条款：{trade}",
            f"运输分类：{klass}",
            f"报价日期：{quote_date}",
        ]
    if (transport_mode or "").strip() == "陆运":
        land_type = (facts.get("运输类型") or "").strip()
        lines = [
            f"工单号：{work_order_no}",
            f"运输方式：{transport_mode}",
            f"运输类型：{land_type or '-'}",
            f"线路类别：{(facts.get('线路类别') or '-').strip() or '-'}",
            f"运输分类：{(facts.get('运输分类') or '-').strip() or '-'}",
        ]
        for key in _land_cargo_keys(land_type, facts):
            label = LAND_DISPLAY.get(key, key)
            val = (facts.get(key) or "-").strip() or "-"
            lines.append(f"{label}：{val}")
        if land_type in _LAND_CUSTOMS_TYPES:
            lines.append(f"通关口岸：{(facts.get('通关口岸') or '').strip()}")
        return lines
    return [
        f"工单号：{work_order_no}",
        f"运输方式：{transport_mode}",
        f"起运地：{origin}",
        f"目的地：{dest}",
        f"品名：{cargo}",
        f"件数：{_with_unit(str(facts.get('件数') or ''), '件')}",
        f"重量（KG）：{_with_unit(str(facts.get('毛重') or ''), 'KGS')}",
        f"体积（CBM）：{_with_unit(str(facts.get('体积') or ''), 'CBM')}",
        f"包装类型：{pack}",
        f"报价日期：{quote_date}",
    ]


def inquiry_card(
    *,
    work_order_no: str,
    facts: dict[str, str],
    first_or_same: str = "first",
    transport_mode: str = "空运",
) -> str:
    """
    建单后的询价确认。

    无标题、无 [test]。first_or_same 保留兼容，不再对销售展示。
    """
    _ = first_or_same
    lines = inquiry_field_lines(
        work_order_no=work_order_no, facts=facts, transport_mode=transport_mode
    )
    lines.append("工单已创建，正在查询TMS报价。")
    return "\n".join(lines)


def canonical_button(raw: str) -> str:
    """把销售点卡 / 原文 / 企微 EventKey 收成标准按钮文案；对不上返回空。"""
    text = (raw or "").strip().strip("【】")
    if text in {
        BTN_SKIP_COLLAB,
        BTN_SKIP_COLLAB_OLD,
        BTN_SKIP_COLLAB_TMS,
        BTN_SKIP_COLLAB_KEY,
    }:
        return BTN_SKIP_COLLAB
    # 新卡 EventKey 带工单号：skip_collab:WO…，避免点旧卡落到最新单
    if text.startswith(f"{BTN_SKIP_COLLAB_KEY}:"):
        return BTN_SKIP_COLLAB
    if text in {BTN_PULL_COLLAB, BTN_PULL_COLLAB_KEY} or text.startswith(
        f"{BTN_PULL_COLLAB_KEY}:"
    ):
        return BTN_PULL_COLLAB
    if text in {BTN_ADOPT_EXCEL, BTN_ADOPT_EXCEL_CARD, BTN_ADOPT_EXCEL_KEY} or text.startswith(
        f"{BTN_ADOPT_EXCEL_KEY}:"
    ):
        return BTN_ADOPT_EXCEL
    if text in {BTN_ADOPT_PDF, BTN_ADOPT_PDF_CARD, BTN_ADOPT_PDF_KEY} or text.startswith(
        f"{BTN_ADOPT_PDF_KEY}:"
    ):
        return BTN_ADOPT_PDF
    if text in {BTN_REJECT_KEY} or text.startswith(f"{BTN_REJECT_KEY}:"):
        return BTN_REJECT
    if text in {BTN_COPY_TICKET, BTN_COPY_TICKET_KEY} or text.startswith(
        f"{BTN_COPY_TICKET_KEY}:"
    ):
        return BTN_COPY_TICKET
    if text in {BTN_DEAL, BTN_DEAL_KEY} or text.startswith(f"{BTN_DEAL_KEY}:"):
        return BTN_DEAL
    if text in {BTN_LOST, BTN_LOST_KEY} or text.startswith(f"{BTN_LOST_KEY}:"):
        return BTN_LOST
    if text in {BTN_NEGOTIATE, BTN_NEGOTIATE_KEY} or text.startswith(
        f"{BTN_NEGOTIATE_KEY}:"
    ):
        return BTN_NEGOTIATE
    if text == BTN_SKIP_COLLAB_DONE:
        return ""
    for btn in (BTN_ADOPT_EXCEL, BTN_ADOPT_PDF, BTN_REJECT, BTN_DEAL, BTN_LOST, BTN_NEGOTIATE):
        if text == btn:
            return btn
    return ""


def extract_work_order_no(text: str) -> str:
    """从任意原话里抽出 WO/VT + 12 位数字；没有则空。不判断意图。"""
    hit = _WO_NO.search(text or "")
    return hit.group(0).upper() if hit else ""


def card_task_id(kind: str, work_order_no: str) -> str:
    """
    每次出卡一个新 task_id。企微 42014：同一 task_id 不能再发，会整卡失败只剩文字。

    工单号仍嵌在前缀里，点卡回调靠 work_order_from_card_meta 找回。
    """
    stamp = f"{int(time.time() * 1000):x}{uuid.uuid4().hex[:4]}"
    return f"{kind}-{work_order_no}-{stamp}"


def looks_like_continue_ticket(text: str) -> bool:
    """
    无网夹具用的弱提示。正式环境必须由大模型判续办意图。
    禁止把本函数当成生产唯一路由。
    """
    raw = (text or "").strip()
    if not raw:
        return False
    return any(token in raw for token in _CONTINUE_HINT)


def work_order_from_card_meta(
    card_meta: dict[str, Any] | None = None,
    *,
    text: str = "",
) -> str:
    """
    从企微点卡回调用找回「点的是哪一张卡当时的工单」。

    优先 TaskId / EventKey 里的工单号（quote-WO…、scheme-adopt_excel-WO…）。
    对不上返回空：调用方按该销售当前最新书签处理。
    """
    src = card_meta or {}
    for raw in (
        str(src.get("task_id") or "").strip(),
        str(src.get("event_key") or "").strip(),
        (text or "").strip(),
    ):
        no = _work_order_in_token(raw)
        if no:
            return no
    return ""


def deal_card_version(card_meta: dict[str, Any] | None = None, *, text: str = "") -> int:
    """从成交卡 EventKey 取出 vN；没有则 0（旧私聊卡）。"""
    src = card_meta or {}
    for raw in (
        str(src.get("event_key") or ""),
        str(src.get("task_id") or ""),
        text or "",
    ):
        hit = re.search(r":v(\d+)\s*$", raw)
        if hit:
            return int(hit.group(1))
    return 0


def _work_order_in_token(raw: str) -> str:
    """从 TaskId / EventKey 里抽出 WO 工单号；没有则空。"""
    text = (raw or "").strip()
    if not text:
        return ""
    for prefix in (
        f"{BTN_SKIP_COLLAB_KEY}:",
        f"{BTN_PULL_COLLAB_KEY}:",
        f"{BTN_ADOPT_EXCEL_KEY}:",
        f"{BTN_ADOPT_PDF_KEY}:",
        f"{BTN_REJECT_KEY}:",
        f"{BTN_COPY_TICKET_KEY}:",
        f"{BTN_DEAL_KEY}:",
        f"{BTN_LOST_KEY}:",
        f"{BTN_NEGOTIATE_KEY}:",
    ):
        if text.startswith(prefix):
            rest = text[len(prefix) :].strip()
            if rest.upper().startswith("WO"):
                return extract_work_order_no(rest) or rest.split("-", 1)[0]
    # 卡片 task_id 会带时间后缀（scheme-WO…-abc），只取 WO+12 位。
    return extract_work_order_no(text)


_CURRENCY_PREFIX = re.compile(r"^[A-Za-z]{3}\b")


def money_text(amount: Any, currency: str = "") -> str:
    """
    销售可见金额：TMS 有币种就写成「CNY 14669.60」。

    金额已带三字码前缀则原样。没有币种不编造 CNY/USD。
    """
    raw = str(amount if amount not in (None, "") else "").strip()
    if not raw or raw == "-":
        return raw or "-"
    if _CURRENCY_PREFIX.match(raw):
        return raw
    code = str(currency or "").strip().upper()
    if code == "RMB":
        code = "CNY"
    if not code:
        return raw
    return f"{code} {raw}"


def _total_label(
    quote: dict[str, Any] | None = None,
    business_line: str = "",
    transport_mode: str = "",
) -> str:
    """销售侧汇总行。企微卡片、文字明细、成交通知都叫费用合计。"""
    src = dict(quote or {})
    return total_price_label(
        business_line or str(src.get("business_line") or ""),
        transport_mode,
        quote=src,
    )


def _manual_quote_lines(
    manual: dict[str, Any],
    *,
    business_line: str = "",
    transport_mode: str = "",
) -> list[str]:
    """人工报价段：费用行 + 总销售价，不写合计。"""
    src = dict(manual or {})
    lines = [f"{name}：{amount}" for name, amount in quote_fee_lines(src)]
    header_ccy = str(src.get("currency") or src.get("currencyCode") or "").strip()
    total = money_text(str(src.get("total") or "-").strip() or "-", header_ccy)
    if total and total != "-":
        lines.append(f"{_total_label(src, business_line, transport_mode)}：{total}")
    elif not lines:
        lines.append(f"{_total_label(src, business_line, transport_mode)}：{total or '-'}")
    return lines


def quote_fee_lines(quote: dict[str, Any] | None) -> list[tuple[str, str]]:
    """
    价格明细行：优先 TMS fee_lines，否则用运费/其它/合计兜底。

    返回 (费用名, 金额) ，空金额不展示。金额带 TMS 币种，不编造。
    """
    src = dict(quote or {})
    header_ccy = str(src.get("currency") or src.get("currencyCode") or "").strip()
    raw = src.get("fee_lines") or src.get("fee_rows")
    parsed: Any = raw
    if isinstance(raw, str) and raw.strip():
        try:
            parsed = json.loads(raw)
        except json.JSONDecodeError:
            parsed = None
    lines: list[tuple[str, str]] = []
    if isinstance(parsed, list):
        for item in parsed:
            if not isinstance(item, dict):
                continue
            name = str(item.get("name") or item.get("label") or "").strip()
            amount = str(item.get("amount") or item.get("value") or "").strip()
            line_ccy = str(item.get("currency") or item.get("currencyCode") or header_ccy).strip()
            if name and amount and amount != "-":
                lines.append((name, money_text(amount, line_ccy)))
    if lines:
        return lines
    freight = str(src.get("freight") or "").strip()
    other = str(src.get("other_fee") or "").strip()
    if freight and freight != "-":
        lines.append(("空运费", money_text(freight, header_ccy)))
    if other and other != "-":
        lines.append(("其他附加费", money_text(other, header_ccy)))
    return lines


_SEA_FEE_BUCKETS = (
    ("polCostItems", "始发港费用", "始发港费用小计"),
    ("oceanCostItems", "海运费", "海运费小计"),
    ("podCostItems", "目的港费用", "目的港费用小计"),
)


def _parse_fee_bucket(quote: dict[str, Any], key: str) -> list[dict[str, Any]]:
    """读 polCostItems / fee_rows_polCostItems，字符串 JSON 也认。"""
    raw = quote.get(key) or quote.get("fee_rows_" + key)
    if isinstance(raw, str) and raw.strip():
        try:
            raw = json.loads(raw)
        except json.JSONDecodeError:
            return []
    if not isinstance(raw, list):
        return []
    return [row for row in raw if isinstance(row, dict)]


def sea_bucket_fee_lines(quote: dict[str, Any] | None) -> list[str]:
    """
    海运三段正文。有任一段桶才用；否则返回空列表，调用方回退扁平 fee_lines。

    费目按该条币种金额原样；小计用 fee_total_* + 表头结算币种。本地不折算。
    """
    src = dict(quote or {})
    header_ccy = str(src.get("currency") or src.get("currencyCode") or "").strip()
    sections: list[str] = []
    any_bucket = False
    for key, title, subtotal_label in _SEA_FEE_BUCKETS:
        rows = _parse_fee_bucket(src, key)
        subtotal = str(src.get("fee_total_" + key) or "").strip()
        if not rows and not subtotal:
            continue
        any_bucket = True
        sections.append(title)
        for row in rows:
            name = str(row.get("name") or row.get("label") or row.get("feeName") or "").strip()
            amount = str(row.get("amount") or row.get("value") or "").strip()
            line_ccy = str(row.get("currency") or row.get("currencyCode") or header_ccy).strip()
            if name and amount and amount != "-":
                sections.append(f"{name}：{money_text(amount, line_ccy)}")
        if subtotal and subtotal != "-":
            sections.append(f"{subtotal_label}：{money_text(subtotal, header_ccy)}")
        sections.append("")
    if not any_bucket:
        return []
    while sections and sections[-1] == "":
        sections.pop()
    return sections


def _land_grouped_fee_lines(quote: dict[str, Any] | None) -> list[str]:
    """陆运：每段先写始发→目的，再写该段 TMS 费用名。"""
    from agent.schema.land_route_options import land_leg_fee_blocks

    src = dict(quote or {})
    header_ccy = str(src.get("currency") or src.get("currencyCode") or "").strip()
    lines: list[str] = []
    for title, fees, eta in land_leg_fee_blocks(src):
        lines.append(title)
        for fee in fees:
            name = str(fee.get("name") or "").strip()
            amount = str(fee.get("amount") or "").strip()
            line_ccy = str(fee.get("currency") or fee.get("currencyCode") or header_ccy).strip()
            if name and amount and amount != "-":
                lines.append(f"{name}：{money_text(amount, line_ccy)}")
        if eta:
            lines.append(f"时效：{eta}")
    return lines


def _is_land_surface(*, quote: dict[str, Any] | None = None, business_line: str = "", transport_mode: str = "") -> bool:
    """陆运销售面：有 landSegments / 陆运业务线就按段出明细。"""
    src = dict(quote or {})
    if src.get("landSegments") or src.get("land_segments") or src.get("selected_land_route"):
        return True
    raw_line = (business_line or "").strip()
    if raw_line and scheme_transport_mode(raw_line, fallback="") == "陆运":
        return True
    mode = (transport_mode or "").strip()
    return bool(mode) and scheme_transport_mode(mode, fallback="") == "陆运"


def quote_area_fee_text(
    quote: dict[str, Any] | None,
    business_line: str = "",
    transport_mode: str = "",
) -> str:
    """
    卡片引用条正文：费用行 + 总销售价。

    必须走 quote_text。塞进 quote_area.title 会被企微整段藏掉。
    """
    src = dict(quote or {})
    header_ccy = str(src.get("currency") or src.get("currencyCode") or "").strip()
    total = money_text(str(src.get("total") or "-").strip() or "-", header_ccy)
    bucket_lines = sea_bucket_fee_lines(src)
    land_lines = _land_grouped_fee_lines(src) if _is_land_surface(
        quote=src, business_line=business_line, transport_mode=transport_mode
    ) else []
    if bucket_lines:
        fee_lines = list(bucket_lines)
    elif land_lines:
        fee_lines = list(land_lines)
    else:
        fee_lines = [f"{name}：{amount}" for name, amount in quote_fee_lines(src)]
    fee_lines.append(f"{_total_label(src, business_line, transport_mode)}：{total}")
    return "\n".join(fee_lines)


# 企微引用条过长会被客户端裁成省略号。海运常有 6～8 项费用，必须走文本明细。
_QUOTE_AREA_SAFE_CHARS = 80


def quote_area_safe_text(
    quote: dict[str, Any] | None,
    business_line: str = "",
    transport_mode: str = "",
) -> str:
    """
    卡片引用条：空运有费用行就只留总价，禁止再塞费用。

    三项短费用也会被企微裁成「人工费：CNY 123...」。完整明细走文本。
    海运/陆运仍按字数：短的原样，超长同样只留总价。
    """
    src = dict(quote or {})
    header_ccy = str(src.get("currency") or src.get("currencyCode") or "").strip()
    total = money_text(str(src.get("total") or "-").strip() or "-", header_ccy)
    n = len(quote_fee_lines(src))
    label = _total_label(src, business_line, transport_mode)
    line = (business_line or "").upper()
    mode = (transport_mode or "").strip()
    is_air = line == "AIR" or mode == "空运"
    if is_air and n:
        return f"{label}：{total}\n共{n}项费用，详见上一条明细"
    full = quote_area_fee_text(quote, business_line=business_line, transport_mode=transport_mode)
    if len(full) <= _QUOTE_AREA_SAFE_CHARS:
        return full
    if n:
        return f"{label}：{total}\n共{n}项费用，详见上一条明细"
    return f"{label}：{total}"


def sea_quote_detail_text(
    *,
    work_order_no: str,
    quote: dict[str, Any],
    first_or_same: str = "first",
    history_work_order_no: str = "",
    status: str = "已报价",
    business_line: str = "SEA",
) -> str:
    """
    完整价格明细文本：来源、工单号、费用行、总价。海运/陆运/空运共用。

    陆运分段时效跟在各段费用后面，用 TMS 原文；报价有效期写在总成本价下面。
    海运时效、报价有效期仍只在按钮卡副标题。
    空运：业务确认没有时效，总销售价下面只留报价有效期。
    不受卡片引用条字数限制。first_or_same 只兼容调用方。
    """
    _ = first_or_same, history_work_order_no, status
    src = dict(quote or {})
    source = str(src.get("source_label") or "TMS标准报价").replace(" ", "")
    lines = [source, f"{work_order_no}价格明细"]
    line = (business_line or "").upper()
    if line == "AIR":
        from agent.schema.air_route_options import route_codes

        route = route_codes(None, {}, src)
        if route:
            lines.append(route)
    if line == "LAND":
        land_line = str(src.get("route_line") or "").strip()
        if land_line:
            lines.append(land_line)
    lines.append(quote_area_fee_text(src, business_line=business_line or "SEA"))
    # 空运、陆运的价格明细正文都要带有效期。海运仍放在按钮卡副标题，避免和原口径打架。
    if line in {"AIR", "LAND"}:
        lines.append(f"报价有效期：{tms_valid_until_display(src)}")
    return "\n".join(lines)


def collab_hint(work_order_no: str) -> str:
    """价格卡协同说明：无「内部协同」前缀，紧挨生成报价单按钮上方。"""
    return (
        f"若需要咨询内部航线人员，请手动拉新人与智能体到原有群，"
        f"并在群聊发送工单编号{work_order_no}。"
    )


def _strip_leading_dash(raw: str) -> str:
    """去掉时效前导 '-' / '—'，避免销售看到「-3-5天」或「-；3-5天」。"""
    text = (raw or "").strip()
    while text.startswith(("-", "—", "–")):
        rest = text[1:].lstrip(" \t；;，,")
        if not rest:
            return ""
        text = rest
    return text


def tms_eta_display(quote: dict[str, Any] | None) -> str:
    """时效只取 TMS 回的 eta / 预计送达；没有就「-」。禁止本地编。前导横杠会剥掉。"""
    src = dict(quote or {})
    eta = _strip_leading_dash(str(src.get("eta") or ""))
    if eta:
        return eta
    dated = _strip_leading_dash(
        str(src.get("eta_date") or src.get("estimated_delivery") or "")
    )
    return dated or "-"


def tms_valid_until_display(quote: dict[str, Any] | None) -> str:
    """报价有效期只取 TMS 原文，没有就 "-"。"""
    src = dict(quote or {})
    raw = str(
        src.get("valid_until")
        or src.get("validUntil")
        or src.get("quoteValidUntil")
        or ""
    ).strip()
    return raw if raw else "-"


def similar_inquiry_line(*, first_or_same: str, history_work_order_no: str = "") -> str:
    """
    相同询价内容检索文案（主账仍会算，销售侧价格卡已隐藏，不再调用）。
    """
    no = (history_work_order_no or "").strip()
    if first_or_same == "same" and no:
        return f"相同询价内容检索：引用历史工单号{no}"
    return "相同询价内容检索：首次询价"


def quote_detail_lines(
    quote: dict[str, Any] | None,
    *,
    first_or_same: str = "first",
    history_work_order_no: str = "",
    status: str = "已报价",
    business_line: str = "",
    transport_mode: str = "",
) -> list[str]:
    """
    价格明细正文：TMS 费用、总销售价一行、有效期；海运/陆运再加时效。

    空运业务确认没有时效，这里不写。禁止编时效/有效期。
    总销售价只写一次。价格卡不展示工单状态，也不展示相同询价内容检索。
    """
    _ = status, first_or_same, history_work_order_no
    src = dict(quote or {})
    header_ccy = str(src.get("currency") or src.get("currencyCode") or "").strip()
    total = money_text(str(src.get("total") or "-").strip() or "-", header_ccy)
    bucket_lines = sea_bucket_fee_lines(src) if (business_line or "").upper() == "SEA" or (
        transport_mode or ""
    ).strip() == "海运" else []
    land_lines = _land_grouped_fee_lines(src) if _is_land_surface(
        quote=src, business_line=business_line, transport_mode=transport_mode
    ) else []
    if bucket_lines:
        lines = list(bucket_lines)
    elif land_lines:
        lines = list(land_lines)
    else:
        lines = [f"{name}：{amount}" for name, amount in quote_fee_lines(src)]
    lines.append(f"{_total_label(src, business_line, transport_mode)}：{total}")
    if not _is_air_surface(business_line=business_line, transport_mode=transport_mode):
        lines.append(f"时效：{tms_eta_display(src)}")
    lines.append(f"报价有效期：{tms_valid_until_display(src)}")
    return lines


def tms_hit_card(
    *,
    work_order_no: str,
    quote: dict[str, Any],
    first_or_same: str = "first",
    history_work_order_no: str = "",
    status: str = "已报价",
) -> str:
    """
    有价后的价格明细（文本兜底，模板卡发失败时仍给销售看）。

    顺序：来源 → 价格明细 → 报价正文 → 协同说明 → 按钮。
    """
    src = dict(quote or {})
    source = str(src.get("source_label") or "TMS标准报价").replace(" ", "")
    lines = [source, f"{work_order_no}价格明细"]
    from agent.schema.air_route_options import route_codes

    route = route_codes(None, {}, src) or str(src.get("route_line") or "").strip()
    if route:
        lines.append(route)
    lines.extend(
        quote_detail_lines(
            src,
            first_or_same=first_or_same,
            history_work_order_no=history_work_order_no,
            status=status,
            business_line="AIR",
        )
    )
    lines.append("")
    lines.append(collab_hint(work_order_no))
    lines.append(BTN_SKIP_COLLAB)
    return "\n".join(lines)


def air_route_entry_card(*, work_order_no: str, count: int, url: str) -> dict[str, Any]:
    """
    多条线路入口卡：短文案跳转 H5，避免企微截断。
    """
    n = max(int(count or 0), 2)
    fallback = "https://ai.ytd-scm.com/inquiry-form"
    action_url = (url or "").strip() or fallback
    return {
        "msgtype": "template_card",
        "template_card": {
            "card_type": "text_notice",
            "source": {"desc": "TMS线路报价", "desc_color": 0},
            "main_title": {"title": f"{work_order_no}已查到{n}条线路"},
            "sub_title_text": "点击查看并选择一条线路，选定后回企微看价格明细。",
            "card_action": {"type": 1, "url": action_url},
            "jump_list": [{"type": 1, "title": "查看并选择线路", "url": action_url}],
        },
    }


def air_route_entry_text(*, work_order_no: str, count: int) -> str:
    """入口卡失败时的文字兜底。"""
    n = max(int(count or 0), 2)
    return f"{work_order_no}已查到{n}条线路，请点击卡片查看并选择。"


def tms_hit_wecom_payload(
    *,
    work_order_no: str,
    quote: dict[str, Any],
    first_or_same: str = "first",
    history_work_order_no: str = "",
    status: str = "已报价",
) -> dict[str, Any]:
    """
    一张价格卡，销售侧顺序：

    价格明细
    ┌ 引用条：空运费 / 其它费用 / 总销售价
    报价有效期（空运不写时效）
    协同说明（普通文字，贴按钮上方）
    按钮

    相同询价内容检索对销售隐藏；first_or_same 只兼容调用方，不展示。
    企微 quote_area.title 只能很短，超长会被整段藏掉。报价必须放 quote_text。
    不用 horizontal：它会排到副标题后面，把费用挤到按钮前。
    """
    _ = first_or_same, history_work_order_no
    src = dict(quote or {})
    source = str(src.get("source_label") or "TMS标准报价").replace(" ", "")
    meta = "\n".join(
        [
            f"报价有效期：{tms_valid_until_display(src)}",
            "",
            collab_hint(work_order_no),
        ]
    )
    return {
        "msgtype": "template_card",
        "template_card": {
            "card_type": "button_interaction",
            "source": {"desc": source, "desc_color": 0},
            "main_title": {"title": f"{work_order_no}价格明细"},
            "quote_area": {
                "type": 0,
                "quote_text": quote_area_safe_text(src, business_line="AIR"),
            },
            "sub_title_text": meta,
            "task_id": card_task_id("quote", work_order_no),
            "button_list": [
                {
                    "text": BTN_SKIP_COLLAB,
                    "style": 1,
                    # EventKey 也带工单号：TaskId 偶发缺失时仍能找回当时那一单
                    "key": f"{BTN_SKIP_COLLAB_KEY}:{work_order_no}",
                }
            ],
        },
    }


def skip_collab_card_update(*, response_code: str, replace_name: str = "") -> dict[str, Any]:
    """
    点过模板卡后更新原卡：被点按钮改成已提交，整卡置灰不可再点。

    必须带企微回调 ResponseCode；无码则不要入出站。
    成交卡三个按钮共用这一条，replace_name 默认「已提交」。
    """
    return {
        "msgtype": "update_template_card",
        "response_code": response_code,
        "button": {"replace_name": (replace_name or BTN_SKIP_COLLAB_DONE).strip() or BTN_SKIP_COLLAB_DONE},
    }


def tms_miss_quote_text(*, work_order_no: str) -> str:
    """文字兜底里的拉群句，与现网图一致。"""
    return f"请手动拉相关人员和智能体到原有群，并在群聊发送工单编号{work_order_no}"


def tms_miss_hint() -> str:
    """卡片标题下灰字：与现网图一致。"""
    return "TMS 标准价库暂无匹配报价"


def tms_miss_body() -> str:
    """卡片中间说明：需求已建单。"""
    return "需求已解析并创建工单。"


def tms_miss(*, work_order_no: str) -> str:
    """
    卡片发送失败时的短兜底。拉群句另发，不写在这里。
    禁止在没打 TMS 前用本句冒充无价。
    """
    no = (work_order_no or "").strip()
    return f"询价工单{no}已建立，TMS标准价库暂无匹配报价。"


def tms_miss_followup(*, work_order_no: str) -> str:
    """出卡后再发的独立一句：拉人进群并带上工单号。"""
    return tms_miss_quote_text(work_order_no=work_order_no)


def copy_ticket_text(*, work_order_no: str) -> str:
    """
    点「复制工单号」后回给销售的正文：只含工单号。

    企微模板卡按钮不能写系统剪贴板。整条消息只有工单号，销售长按「复制」不会夹杂其它字。
    """
    return (work_order_no or "").strip()


def tms_miss_wecom_payload(
    *,
    work_order_no: str,
    action_url: str = "",
    copy_url: str = "",
) -> dict[str, Any]:
    """
    TMS 已查过、确认没价：一张文本通知卡（对齐现网图）。

    询价工单号 / 灰字无价说明 / 建单说明 / 工单状态与查价结果。
    不要「来源」行、不要按钮。企微 text_notice 必须带可信域名 card_action。
    copy_url 忽略：不再跳复制页。
    """
    _ = copy_url
    no = (work_order_no or "").strip()
    card: dict[str, Any] = {
        "card_type": "text_notice",
        "source": {"desc": "TMS查价结果", "desc_color": 0},
        "main_title": {
            "title": f"询价工单 {no}",
            "desc": tms_miss_hint(),
        },
        "sub_title_text": tms_miss_body(),
        "horizontal_content_list": [
            {"keyname": "工单状态", "value": "询价中"},
            {"keyname": "查价结果", "value": "暂无匹配报价"},
        ],
        "task_id": card_task_id("miss", no),
    }
    url = (action_url or "").strip()
    if url:
        card["card_action"] = {"type": 1, "url": url}
    return {"msgtype": "template_card", "template_card": card}


def tms_contract_gap(*, work_order_no: str) -> str:
    """已废弃：不再因字典对不上拦查价；保留以免旧单测/引用报错。"""
    return tms_miss(work_order_no=work_order_no)


def tms_need_more(*, work_order_no: str, hint: str = "") -> str:
    """已废弃：TMS 回 1003 按合同收成暂无报价。保留以免旧引用报错。"""
    _ = hint
    return tms_miss(work_order_no=work_order_no)


def skip_collab_ack() -> str:
    return "好的，已跳过内部协同。正在按 TMS 标准报价生成报价单草稿…"


def sea_collab_ask() -> str:
    """海运有价卡协同问句。"""
    return "是否拉产品人员进群协同？"


def sea_miss_ask() -> str:
    """海运无价卡协同问句。"""
    return "未查到合适报价，是否需要拉内部产品人员进群，进行手工协同报价？"


def sea_tms_hit_card(
    *,
    work_order_no: str,
    quote: dict[str, Any],
    first_or_same: str = "first",
    history_work_order_no: str = "",
    status: str = "已报价",
    business_line: str = "SEA",
) -> str:
    """海运/陆运有价：价格明细 + 拉群问句 + 两个按钮。空运卡片函数保持不动。"""
    src = dict(quote or {})
    source = str(src.get("source_label") or "TMS标准报价").replace(" ", "")
    lines = [source, f"{work_order_no}价格明细"]
    lines.extend(
        quote_detail_lines(
            src,
            first_or_same=first_or_same,
            history_work_order_no=history_work_order_no,
            status=status,
            business_line=business_line or "SEA",
        )
    )
    lines.append("")
    lines.append(sea_collab_ask())
    lines.append(BTN_PULL_COLLAB)
    lines.append(BTN_SKIP_COLLAB_TMS)
    return "\n".join(lines)


def sea_tms_hit_wecom_payload(
    *,
    work_order_no: str,
    quote: dict[str, Any],
    first_or_same: str = "first",
    history_work_order_no: str = "",
    status: str = "已报价",
    business_line: str = "SEA",
) -> dict[str, Any]:
    """海运/陆运有价企微卡：价格明细 + 问句 + 拉群 / 跳过协同。"""
    _ = first_or_same, history_work_order_no, status
    src = dict(quote or {})
    source = str(src.get("source_label") or "TMS标准报价").replace(" ", "")
    meta = "\n".join(
        [
            f"时效：{tms_eta_display(src)}",
            f"报价有效期：{tms_valid_until_display(src)}",
            "",
            sea_collab_ask(),
        ]
    )
    return {
        "msgtype": "template_card",
        "template_card": {
            "card_type": "button_interaction",
            "source": {"desc": source, "desc_color": 0},
            "main_title": {"title": f"{work_order_no}价格明细"},
            "quote_area": {
                "type": 0,
                "quote_text": quote_area_safe_text(src, business_line=business_line or "SEA"),
            },
            "sub_title_text": meta,
            "task_id": card_task_id("quote", work_order_no),
            "button_list": [
                {
                    "text": BTN_PULL_COLLAB,
                    "style": 1,
                    "key": f"{BTN_PULL_COLLAB_KEY}:{work_order_no}",
                },
                {
                    "text": BTN_SKIP_COLLAB_TMS,
                    "style": 1,
                    "key": f"{BTN_SKIP_COLLAB_KEY}:{work_order_no}",
                },
            ],
        },
    }


def sea_tms_miss(*, work_order_no: str) -> str:
    """海运无价文字兜底：工单号、暂无匹配、询价中、拉群问句。"""
    no = (work_order_no or "").strip()
    return "\n".join(
        [
            f"询价工单 {no}",
            "暂无匹配报价",
            "工单状态：询价中",
            sea_miss_ask(),
            BTN_PULL_COLLAB,
        ]
    )


def sea_tms_miss_wecom_payload(*, work_order_no: str, action_url: str = "") -> dict[str, Any]:
    """
    海运无价一张卡：工单号、暂无匹配、询价中、拉群问句、一个按钮。

    禁止再写「请手动拉人到原有群」。
    """
    no = (work_order_no or "").strip()
    card: dict[str, Any] = {
        "card_type": "button_interaction",
        "source": {"desc": "TMS查价结果", "desc_color": 0},
        "main_title": {
            "title": f"询价工单 {no}",
            "desc": "暂无匹配报价",
        },
        "sub_title_text": "\n".join(
            [
                "需求已解析并创建工单。",
                sea_miss_ask(),
            ]
        ),
        "horizontal_content_list": [
            {"keyname": "工单状态", "value": "询价中"},
            {"keyname": "查价结果", "value": "暂无匹配报价"},
        ],
        "task_id": card_task_id("miss", no),
        "button_list": [
            {
                "text": BTN_PULL_COLLAB,
                "style": 1,
                "key": f"{BTN_PULL_COLLAB_KEY}:{no}",
            }
        ],
    }
    url = (action_url or "").strip()
    if url:
        card["card_action"] = {"type": 1, "url": url}
    return {"msgtype": "template_card", "template_card": card}


MULTI_CONFIRM_TAIL = "请回复「确定」创建工单，内容有误时，请重新发内容。"
MULTI_SAME_MODE = "同一运输方式出现了两段，请拆开再发。"
MULTI_CONFIRM_EXPIRED = "刚才的多段核对已过期，请把多段需求再发一次，出核对后再回复确定。"


def multi_air_manual_hint(*, work_order_no: str) -> str:
    """含空运的多段：不自动拉群，请销售把人和智能体拉进原有群并发送工单号。"""
    no = (work_order_no or "").strip()
    return f"请手动拉相关人员和智能体到原有群，并在群聊发送工单编号{no}"


def multi_pull_text(*, work_order_no: str) -> str:
    """不含空运：建单后的拉群文字兜底。不写 TMS、不写暂无报价。"""
    no = (work_order_no or "").strip()
    return "\n".join(
        [
            f"询价工单 {no}",
            "工单已创建。",
            "是否拉产品人员进群协同？",
            BTN_PULL_COLLAB,
        ]
    )


def multi_pull_wecom_payload(*, work_order_no: str) -> dict[str, Any]:
    """不含空运：一张卡、一个按钮，点下去才按各段线路建群。"""
    no = (work_order_no or "").strip()
    return {
        "msgtype": "template_card",
        "template_card": {
            "card_type": "button_interaction",
            "source": {"desc": "多段联运", "desc_color": 0},
            "main_title": {"title": f"询价工单 {no}", "desc": "工单已创建。"},
            "sub_title_text": "是否拉产品人员进群协同？",
            "task_id": card_task_id("multi_pull", no or "multi"),
            "button_list": [
                {
                    "text": BTN_PULL_COLLAB,
                    "style": 1,
                    "key": f"{BTN_PULL_COLLAB_KEY}:{no}",
                }
            ],
        },
    }


def handoff_air_hint(*, work_order_no: str) -> str:
    """空运转人工：只发文字，不建群。"""
    no = (work_order_no or "").strip()
    if no:
        return f"触发转人工，请手动把航线和智能体拉进原有空运群，并发送工单号 {no}。"
    return "触发转人工，请手动把航线和智能体拉进原有空运群，并发送工单号。"


def handoff_offer_text(*, reason: str, transport_mode: str) -> str:
    """海运/陆运转人工卡片的文字兜底（点卡或原样回复按钮均可）。"""
    why = (reason or "").strip() or HANDOFF_INTENT_REASON
    mode = (transport_mode or "").strip()
    lines = [why]
    if mode:
        lines.append(f"运输方式：{mode}")
    lines.append("是否拉产品人员进群协同？")
    lines.append(BTN_PULL_COLLAB)
    return "\n".join(lines)


def handoff_offer_wecom_payload(*, reason: str, task_token: str) -> dict[str, Any]:
    """海运/陆运一张卡、一个按钮：拉产品进群协同。"""
    token = (task_token or "handoff").strip() or "handoff"
    why = (reason or "").strip() or HANDOFF_INTENT_REASON
    return {
        "msgtype": "template_card",
        "template_card": {
            "card_type": "button_interaction",
            "source": {"desc": "转人工", "desc_color": 0},
            "main_title": {"title": why, "desc": "是否拉产品人员进群协同？"},
            "task_id": card_task_id("handoff", token),
            "button_list": [
                {
                    "text": BTN_PULL_COLLAB,
                    "style": 1,
                    "key": f"{BTN_PULL_COLLAB_KEY}:{token}",
                }
            ],
        },
    }


def handoff_group_brief(
    *,
    work_order_no: str,
    facts: dict[str, str] | None,
    business_line: str,
    product_names: list[str] | None = None,
    sales_name: str = "",
    airline_names: list[str] | None = None,
) -> str:
    """
    转人工群摘要：@人 +「工单号转人工」+ 已整理询价；未整理与协同全部为 -。
    不写 TMS、不写模板。
    """
    from agent.policy.handoff_human import collab_keys_for_handoff, transport_label
    from agent.schema.field_validate import collab_display_name

    no = (work_order_no or "").strip()
    line = (business_line or "").strip().upper()
    mode = transport_label(line) or "海运"
    names: list[str] = []
    if line == "AIR":
        if (sales_name or "").strip():
            names.append(sales_name.strip())
        for n in airline_names or []:
            if (n or "").strip() and n.strip() not in names:
                names.append(n.strip())
    else:
        for n in product_names or []:
            if (n or "").strip() and n.strip() not in names:
                names.append(n.strip())
    at = " ".join(f"@{n}" for n in names)
    src = dict(facts or {})
    field_lines = inquiry_field_lines(work_order_no=no, facts=src, transport_mode=mode)
    body = field_lines[1:] if field_lines else []
    lines: list[str] = []
    if at:
        lines.append(at)
    lines.append(f"{no}转人工")
    lines.append("询价字段：")
    lines.extend(body)
    lines.append("")
    lines.append("协同补充字段：")
    for key in collab_keys_for_handoff(src, line):
        label = collab_display_name(key, line)
        lines.append(f"{label}：-")
    return "\n".join(lines)


def _role_label(name: str, role: str) -> str:
    """姓名前加岗位，避免「销售甲、乙」被读成两人都是销售。已带岗位前缀的不再重复加。"""
    text = (name or "").strip()
    if not text:
        return role
    if text.startswith(role):
        return text
    return f"{role}{text}"


def sea_group_created(*, work_order_no: str, sales_name: str, product_names: list[str]) -> str:
    """拉群成功后的私聊。已邀请写成「销售某某、产品某某」，岗位不能省略。"""
    no = (work_order_no or "").strip()
    sales = _role_label(sales_name, "销售")
    products = "、".join(
        _role_label(n, "产品") for n in (product_names or []) if (n or "").strip()
    ) or "产品"
    return (
        f"协同群已创建，群名与工单号一致：{no}。"
        f"已邀请：{sales}、{products}。请在群聊中继续确认报价。"
    )


def sea_group_exists(*, work_order_no: str) -> str:
    """同一工单已有群，不再新建。"""
    no = (work_order_no or "").strip()
    return f"协同群已在，群名与工单号一致：{no}。请到该群继续确认报价。"


def sea_group_no_staff() -> str:
    """没有可拉的产品人员，不建空群。"""
    return "没有可拉的产品人员。请先在后台「员工管理」维护本线路的海运产品岗后再试。"


def group_sheet_send_fail() -> str:
    """模板已填好但没发进群。"""
    return "报价单已生成，发到群里失败，请稍后重试。"


def sea_group_create_fail(*, reason: str = "") -> str:
    """建群失败，按钮保持可点。"""
    extra = (reason or "").strip()
    if extra:
        return f"建群失败：{extra}。请稍后重试。"
    return "建群失败，请稍后重试。"


def quote_is_product(quote: dict[str, Any] | None) -> bool:
    """来源是产品/航线人工，不能写进 TMS 报价段。"""
    src = dict(quote or {})
    code = str(src.get("source") or "").strip().upper()
    if code == "PRODUCT":
        return True
    if code in {"TMS", "SALES_ADJUST"}:
        return False
    label = str(src.get("source_label") or src.get("sourceLabel") or src.get("source") or "")
    return "产品" in label or "航线" in label


def quote_is_sales_adjust(quote: dict[str, Any] | None) -> bool:
    """销售调过价的这一版，禁止再写进 TMS 报价段。"""
    src = dict(quote or {})
    code = str(src.get("source") or "").strip().upper()
    if code == "SALES_ADJUST":
        return True
    if code in {"TMS", "PRODUCT"}:
        return False
    label = str(src.get("source_label") or src.get("sourceLabel") or "").replace(" ", "")
    return "销售调价" in label


def quote_is_tms(quote: dict[str, Any] | None) -> bool:
    """
    只有 TMS 原价能进 TMS 段。

    产品报价、销售调价即使金额还在，也不能冒充 TMS。
    """
    src = dict(quote or {})
    if not src:
        return False
    if quote_is_product(src) or quote_is_sales_adjust(src):
        return False
    code = str(src.get("source") or "").strip().upper()
    if code == "TMS":
        return True
    label = str(src.get("source_label") or src.get("sourceLabel") or src.get("source") or "")
    return "TMS" in label.upper() or "标准报价" in label or not label.strip()


def remember_tms_quote(sess: Any, quote: dict[str, Any] | None) -> None:
    """
    书签里单独记住 TMS 原价。调价只改 sess.quote，不准覆盖这一份。

    副作用：写 sess.tms_quote。调用方在 TMS 命中或调价前调用。
    """
    if sess is None:
        return
    src = dict(quote or {})
    if not quote_is_tms(src):
        return
    sess.tms_quote = dict(src)


def resolve_tms_quote(sess: Any = None, ticket: Any = None) -> dict[str, Any]:
    """
    拉群/群摘要用的 TMS 原价：书签 tms_quote → 主账 tms_quote → 当前价若仍是 TMS。

    调后最新一条是销售调价，不能拿来当 TMS。
    """
    candidates: list[dict[str, Any]] = []
    if sess is not None:
        candidates.append(dict(getattr(sess, "tms_quote", None) or {}))
    view = dict(ticket or {}) if isinstance(ticket, dict) else {}
    if view:
        candidates.append(dict(view.get("tms_quote") or view.get("tmsQuote") or {}))
    elif ticket is not None:
        candidates.append(dict(getattr(ticket, "tms_quote", None) or {}))
    if sess is not None:
        candidates.append(dict(getattr(sess, "quote", None) or {}))
    if view:
        candidates.append(dict(view.get("quote") or {}))
    for item in candidates:
        if quote_is_tms(item) and (item.get("total") or item.get("fee_rows") or item.get("fee_lines")):
            return item
    return {}


def _land_quote_is_selected(quote: dict[str, Any] | None) -> bool:
    """已收成一条陆运线路：有选定段，或只剩一条且写了线路名。"""
    src = dict(quote or {})
    selected = src.get("selected_land_route")
    if isinstance(selected, dict) and list(selected.get("legs") or []):
        return True
    if not str(src.get("route_line") or "").strip():
        return False
    from agent.schema.land_route_options import needs_route_pick

    return not needs_route_pick(src)


def land_group_tms_quote(
    tms: dict[str, Any] | None,
    current: dict[str, Any] | None,
) -> dict[str, Any]:
    """
    群摘要陆运 TMS 段用已选线路，不用还没选的整包。

    选定前书签里的 TMS 原价含全部备选费用，总成本价和时效都还没汇总。
    当前价已是选定投影时用当前价。当前价已是人工报价或调价时，仍用传入的 TMS 价。
    """
    raw = dict(tms or {})
    cur = dict(current or {})
    if not quote_is_tms(cur) or not _land_quote_is_selected(cur):
        return raw
    from agent.schema.land_route_options import needs_route_pick

    if _land_quote_is_selected(raw) and not needs_route_pick(raw):
        return raw
    return cur


def sea_group_brief(
    *,
    work_order_no: str,
    facts: dict[str, str],
    quote: dict[str, Any] | None,
    collab_facts: dict[str, str] | None,
    sales_name: str,
    product_names: list[str],
    has_price: bool,
    tms_quote: dict[str, Any] | None = None,
    product_quote: dict[str, Any] | None = None,
    transport_mode: str = "海运",
) -> str:
    """
    群里自动摘要：询价 + 协同 + TMS 报价 + 人工报价。

    TMS 无价写「TMS 暂无匹配报价」并提示用文字或 Excel 另报；产品报的价单独放人工报价，销售调价另起一段。
    禁止把调后价写进 TMS 段。陆运传入 transport_mode=陆运，始发站/货物字段跟核对卡同一套。
    """
    no = (work_order_no or "").strip()
    sales = (sales_name or "").strip() or "销售"
    products = [n.strip() for n in (product_names or []) if (n or "").strip()]
    at_bits = [f"@{sales}"] + [f"@{n}" for n in products]
    mode = (transport_mode or "海运").strip() or "海运"
    lines = [" ".join(at_bits), f"工单号：{no}", "询价字段："]
    lines.extend(
        inquiry_field_lines(work_order_no=no, facts=facts, transport_mode=mode)[1:]
    )
    lines.append("")
    lines.append("协同补充字段：")
    collab = dict(collab_facts or {})
    line = "LAND" if mode == "陆运" else "SEA"
    collab_keys = collab_fields_for_ticket(facts, line) or SEA_COLLAB_FIELDS
    for key in collab_keys:
        label = collab_display_name(key, line)
        val = (collab.get(key) or collab.get(label) or "").strip() or "-"
        lines.append(f"{label}：{val}")
    tms, manual, adjust = _split_brief_quotes(
        quote=quote,
        tms_quote=tms_quote,
        product_quote=product_quote,
        has_price=has_price,
    )
    if mode == "陆运":
        # 拉群时书签仍可能是整包。当前价已选定时换成那一条，和私聊明细对齐。
        tms = land_group_tms_quote(tms, quote)
    lines.append("")
    lines.append("TMS报价：")
    if tms and (tms.get("total") or quote_fee_lines(tms)):
        if mode == "陆运":
            # 与私聊价格明细同一套：线路名、分段费用、各段时效、总成本价、有效期。
            # 不用 quote_detail_lines，那会再补一行合并时效，也会在未选定时摊开全部费用。
            # 群里已有「TMS报价：」，不要再叠「TMS标准报价」和「工单号价格明细」。
            detail = sea_quote_detail_text(
                work_order_no=no,
                quote=tms,
                business_line="LAND",
            ).splitlines()
            source = str(tms.get("source_label") or "TMS标准报价").replace(" ", "")
            title = f"{no}价格明细"
            lines.extend(ln for ln in detail if ln.strip() not in {source, title, "TMS标准报价"})
        else:
            lines.extend(quote_detail_lines(tms, transport_mode=mode))
        lines.append(group_tms_confirm_hint(business_line="LAND" if mode == "陆运" else "SEA"))
    else:
        lines.append("TMS 暂无匹配报价")
        lines.append(group_no_tms_quote_hint(business_line="LAND" if mode == "陆运" else "SEA"))
    if manual and (quote_fee_lines(manual) or str(manual.get("total") or "").strip() not in {"", "-"}):
        lines.append("")
        lines.append("人工报价：")
        lines.extend(_manual_quote_lines(manual, transport_mode=mode))
    if adjust and (quote_fee_lines(adjust) or str(adjust.get("total") or "").strip() not in {"", "-"}):
        lines.append("")
        lines.append("销售调价：")
        lines.extend(_manual_quote_lines(adjust, transport_mode=mode))
    return "\n".join(lines)


def _split_brief_quotes(
    *,
    quote: dict[str, Any] | None,
    tms_quote: dict[str, Any] | None,
    product_quote: dict[str, Any] | None,
    has_price: bool,
) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
    """
    群摘要三段价：TMS 原价 / 人工 / 销售调价。调后价不准掉进 TMS。

    tms_quote 显式传入（含空字典）时不再用当前 quote 兜底。
    """
    src = dict(quote or {})
    tms = dict(tms_quote) if tms_quote is not None else {}
    manual = dict(product_quote or {})
    if tms_quote is None:
        if has_price and src and quote_is_tms(src):
            tms = src
        elif quote_is_product(src):
            manual = manual or src
    if quote_is_sales_adjust(tms) or quote_is_product(tms):
        tms = {}
    if quote_is_sales_adjust(manual):
        manual = {}
    adjust = src if quote_is_sales_adjust(src) else {}
    if quote_is_sales_adjust(product_quote) and not adjust:
        adjust = dict(product_quote or {})
    return tms, manual, adjust


def scheme_card(
    *,
    work_order_no: str,
    status: str,
    facts: dict[str, str],
    quote: dict[str, Any],
    transport_mode: str = "空运",
) -> str:
    """
    报价方案文本兜底（模板卡发失败时仍给销售看）。

    与卡片同一套字段，禁止漏项。不再展示工单状态。
    """
    _ = status
    src = dict(quote or {})
    # 方案卡不展示询价「报价日期」，改展示 TMS 报价有效期
    body = "\n".join(
        line
        for line in inquiry_field_lines(
            work_order_no=work_order_no, facts=facts, transport_mode=transport_mode
        )
        if not line.startswith("报价日期：")
    )
    eta = tms_eta_display(src)
    valid = tms_valid_until_display(src)
    lines = [
        f"【{work_order_no}报价方案】",
        body,
        "价格明细",
        quote_area_fee_text(src, transport_mode=transport_mode),
    ]
    if not _is_air_surface(transport_mode=transport_mode):
        lines.append(f"时效：{eta or '-'}")
    lines.extend(
        [
            f"报价有效期：{valid}",
            "请确认是否采用该报价方案：",
            f"【{BTN_ADOPT_EXCEL_CARD}】",
            f"【{BTN_ADOPT_PDF_CARD}】",
            f"【{BTN_REJECT}】",
        ]
    )
    return "\n".join(lines)


def _clip(raw: str, limit: int) -> str:
    """企微列表格有字数建议；超长才裁，避免空白。"""
    text = (raw or "").strip() or "-"
    return text if len(text) <= limit else text[: limit - 1] + "…"


def _air_scheme_route(quote: dict[str, Any], facts: dict[str, str]) -> str:
    """方案卡航线：空运ZUH→中转→CRK。没有三字码才退回地名。"""
    from agent.schema.air_route_options import scheme_route_codes

    return scheme_route_codes(quote, facts)


def scheme_transport_mode(business_line: str = "", fallback: str = "空运") -> str:
    """会话业务线 → 方案卡运输方式。海运不得再写成空运。"""
    raw = (business_line or "").strip()
    line = raw.upper()
    if line == "SEA" or raw == "海" or "海运" in raw:
        return "海运"
    if line == "LAND" or raw == "陆" or "陆运" in raw:
        return "陆运"
    if line == "AIR" or raw == "空" or "空运" in raw:
        return "空运"
    return (fallback or "空运").strip() or "空运"


def _is_air_surface(*, business_line: str = "", transport_mode: str = "") -> bool:
    """
    空运销售面：业务确认没有时效，展示层一律不写。

    海运/陆运仍保留时效。两边都空时不算空运，避免误藏海运/陆运卡。
    """
    raw_line = (business_line or "").strip()
    if raw_line and scheme_transport_mode(raw_line, fallback="") == "空运":
        return True
    mode = (transport_mode or "").strip()
    return bool(mode) and scheme_transport_mode(mode, fallback="") == "空运"


def scheme_horizontal_rows(
    *,
    work_order_no: str,
    facts: dict[str, str],
    quote: dict[str, Any],
    transport_mode: str = "空运",
) -> list[dict[str, str]]:
    """
    卡片字段：一行一项，每格独立字数，不会像副标题整段截成 8CB…。

    企微 horizontal 最多 6 条。运输方式写进行程行，避免丢字段。
    海运用货物数量/整柜或拼柜/箱型箱量，禁止套空运件数毛重体积包装。
    陆运用运输类型/线路类别/体积重量或车型数量，禁止套海运箱型或空运件数包装。
    不用 sub_title 堆字段（它会排到列表前面，且超长必截断）。
    """
    src = facts or {}
    q = dict(quote or {})
    origin = (src.get("起运港") or "-").strip() or "-"
    dest = (src.get("目的港") or "-").strip() or "-"
    cargo = (src.get("品名") or "-").strip() or "-"
    mode = (transport_mode or "空运").strip() or "空运"
    if mode == "海运":
        qty = (src.get("货量") or src.get("货物数量") or "-").strip() or "-"
        load = (src.get("整柜或拼柜") or "-").strip() or "-"
        box = (src.get("箱型箱量") or "-").strip() or "-"
        return [
            {"keyname": "工单号", "value": _clip(work_order_no, 26)},
            {"keyname": "航线", "value": _clip(f"海运 {origin}→{dest}", 26)},
            {"keyname": "货物", "value": _clip(f"{cargo}；{qty}", 26)},
            {"keyname": "箱型", "value": _clip(f"{load}；{box}", 26)},
            {"keyname": "时效", "value": _clip(tms_eta_display(q), 26)},
            {"keyname": "报价有效期", "value": _clip(tms_valid_until_display(q), 26)},
        ]
    if mode == "陆运":
        return [
            {"keyname": "工单号", "value": _clip(work_order_no, 26)},
            {"keyname": "线路", "value": _clip(f"陆运 {origin}→{dest}", 26)},
            {"keyname": "类型", "value": _clip(_land_scheme_type_line(src), 26)},
            {"keyname": "货物", "value": _clip(_land_scheme_goods_line(src), 26)},
            {"keyname": "时效", "value": _clip(tms_eta_display(q), 26)},
            {"keyname": "报价有效期", "value": _clip(tms_valid_until_display(q), 26)},
        ]
    pieces = _with_unit(str(src.get("件数") or ""), "件")
    weight = _with_unit(str(src.get("毛重") or ""), "KGS")
    volume = _with_unit(str(src.get("体积") or ""), "CBM")
    pack = (src.get("包装方式") or src.get("包装类型") or "-").strip() or "-"
    return [
        {"keyname": "工单号", "value": _clip(work_order_no, 26)},
        {"keyname": "航线", "value": _clip(_air_scheme_route(q, src), 26)},
        {"keyname": "货物", "value": _clip(f"{cargo}；{pieces}；{weight}；{volume}", 26)},
        {"keyname": "包装", "value": _clip(pack, 26)},
        {"keyname": "报价有效期", "value": _clip(tms_valid_until_display(q), 26)},
    ]


def _land_scheme_type_line(facts: dict[str, str]) -> str:
    """方案卡「类型」格：运输类型、线路类别、运输分类。"""
    src = facts or {}
    land_type = (src.get("运输类型") or "-").strip() or "-"
    bits = [land_type]
    route = (src.get("线路类别") or "-").strip() or "-"
    klass = (src.get("运输分类") or "-").strip() or "-"
    bits.extend([route, klass])
    return "；".join(bits)


def _land_scheme_goods_line(facts: dict[str, str]) -> str:
    """方案卡「货物」格：品名 + 当前类型的体积重量或车型/数量。"""
    src = facts or {}
    land_type = (src.get("运输类型") or "").strip()
    bits: list[str] = [(src.get("品名") or "-").strip() or "-"]
    for key in _land_cargo_keys(land_type, src):
        if key in {"起运港", "目的港", "品名", "报价日期"}:
            continue
        val = (src.get(key) or "").strip()
        if val:
            bits.append(val)
    if land_type in _LAND_CUSTOMS_TYPES:
        customs = (src.get("通关口岸") or "").strip()
        if customs:
            bits.append(customs)
    return "；".join(bits)


def scheme_wecom_payload(
    *,
    work_order_no: str,
    facts: dict[str, str],
    quote: dict[str, Any],
    status: str = "已报价",
    transport_mode: str = "空运",
) -> dict[str, Any]:
    """
    一张报价方案卡：标题带工单号、引用条、字段、三个短按钮。

    按钮文案必须短：生成Excel / 生成PDF / 不采用。长文案横排会被裁。
    海运/空运/陆运字段分开，禁止陆运卡再出箱型或空运件数包装。
    """
    _ = status
    src = dict(quote or {})
    source = str(src.get("source_label") or "TMS标准报价").replace(" ", "")
    return {
        "msgtype": "template_card",
        "template_card": {
            "card_type": "button_interaction",
            "source": {"desc": source, "desc_color": 0},
            "main_title": {"title": f"{work_order_no}报价方案"},
            "quote_area": {
                "type": 0,
                # 空运/海运/陆运引用条一律只留总价，费用在上一条明细。
                "quote_text": quote_area_safe_text(src, transport_mode=transport_mode),
            },
            "horizontal_content_list": scheme_horizontal_rows(
                work_order_no=work_order_no,
                facts=facts,
                quote=src,
                transport_mode=transport_mode,
            ),
            "task_id": card_task_id("scheme", work_order_no),
            "button_list": [
                {
                    "text": BTN_ADOPT_EXCEL_CARD,
                    "style": 1,
                    "key": f"{BTN_ADOPT_EXCEL_KEY}:{work_order_no}",
                },
                {
                    "text": BTN_ADOPT_PDF_CARD,
                    "style": 1,
                    "key": f"{BTN_ADOPT_PDF_KEY}:{work_order_no}",
                },
                {
                    "text": BTN_REJECT,
                    "style": 2,
                    "key": f"{BTN_REJECT_KEY}:{work_order_no}",
                },
            ],
        },
    }


def reject_ack() -> str:
    return "请问如何调整报价单，以满足您的需求？请说明要调哪一项费用，例如「提货费加 10%」。"


def adjust_confirm(applied: list[dict[str, Any]]) -> str:
    """识别成功确认。applied 每项含 name、how。"""
    bits = [f"{x.get('name')}在当前报价基础上{x.get('how')}" for x in applied if x.get("name")]
    body = "；".join(bits) or "已识别调价指令"
    return f"已识别调价指令：{body}。正在刷新报价"


def adjust_ledger_remark(applied: list[dict[str, Any]]) -> str:
    """
    主账状态流转备注：每次销售调价一条，记下当时改了哪几项。

    后台「状态流转与备注」原样展示；不含利润点。
    """
    bits = [f"{x.get('name')}{x.get('how')}" for x in applied if x.get("name")]
    return "销售调价：" + ("；".join(bits) if bits else "已改价")


def adjust_need_fee() -> str:
    return "请说明要调哪一项费用明细，例如「提货费加 10%」。只说报价或总价，系统不会改价。"


def adjust_need_lines() -> str:
    return "请分别对明细价格进行调价，系统不分摊合计。"


def adjust_currency_hint() -> str:
    return "请分别对不同币种进行调价，系统不做汇率换算。"


def adjust_unmatched(missing: list[str], available: list[str]) -> str:
    miss = "、".join(x for x in missing if x) or "这些费用"
    have = "、".join(available) if available else "（当前没有费用明细）"
    return f"卡上没有{miss}。当前能调的费用：{have}"


def adjust_ambiguous(name: str, hits: list[str]) -> str:
    return f"「{name}」对上了多项（{'、'.join(hits)}），请说明要调哪一项。"


def adjust_ineligible(reason: str) -> str:
    if reason == "already_collab":
        return "这张单已经拉群协同，请到群里发送调价指令。"
    if reason == "closed":
        return "这张单已经结案，不能再按这套调价。"
    return "当前工单还没有可调的报价，不能按这套调价。"


def adjust_only_sales() -> str:
    return "只有这张单的销售能调价。航线/产品请继续报价，价格未改。"


def adjust_group_body(quote: dict[str, Any] | None, business_line: str = "") -> str:
    """
    群里调后全量明细，利润点不写给销售。

    海运必须带三段标题与小计（始发港/海运/目的港），禁止只摊平 fee_lines 把小计弄丢。
    """
    src = dict(quote or {})
    text = quote_area_fee_text(src, business_line=business_line)
    return "调价后价格：\n" + text


def adopt_excel_ack() -> str:
    return "已确认采用该方案，正在生成 Excel 最终报价单…"


def adopt_pdf_ack() -> str:
    return "已确认采用该方案，正在生成PDF最终报价单…"


QUOTE_TEMPLATE_MISSING = "生成报价单失败，请联系IT运维维护后台报价模板。"


def quote_template_missing() -> str:
    """关键词与默认模板都未维护时的销售提示（产品口径，勿改字）。"""
    return QUOTE_TEMPLATE_MISSING


def file_ready(*, kind: str, work_order_no: str, filename: str) -> str:
    return f"已生成{kind}最终报价单：{filename}（工单 {work_order_no}）"


# 未成交原因说明跟在未成交后，不跟在协商中后。验收脚本认这个开关。
DEAL_LOST_REASON_BEFORE_NEGOTIATE = True
DEAL_HINT = "咨询成交状态：请输入成交 / 未成交 (未成交需补充原因)/ 协商中。"
DEAL_SUBTITLE = "报价单已发送，请跟进成交状态"


def deal_horizontal_rows(
    *,
    quote: dict[str, Any] | None,
    facts: dict[str, str] | None = None,
    business_line: str = "",
    transport_mode: str = "",
) -> list[dict[str, str]]:
    """
    成交跟进卡字段：费用合计、报价有效期；海运/陆运再加时效。

    空运业务确认没有时效，这里不写。
    不要历史报价 / 工单状态 / 来源。来源只留卡片顶部红字。
    """
    _ = facts
    src = dict(quote or {})
    header_ccy = str(src.get("currency") or src.get("currencyCode") or "").strip()
    total = money_text(str(src.get("total") or "-").strip() or "-", header_ccy)
    rows = [
        {"keyname": _total_label(src, business_line, transport_mode), "value": _clip(total, 26)},
    ]
    if not _is_air_surface(business_line=business_line, transport_mode=transport_mode):
        rows.append({"keyname": "时效", "value": _clip(tms_eta_display(src), 26)})
    rows.append({"keyname": "报价有效期", "value": _clip(tms_valid_until_display(src), 26)})
    return rows


def deal_card(
    *,
    work_order_no: str,
    status: str = "",
    summary: str = "",
    quote: dict[str, Any] | None = None,
    facts: dict[str, str] | None = None,
    business_line: str = "",
) -> str:
    """
    成交跟进文字兜底（模板卡发失败时仍给销售看）。

    与卡片同一套字段；status / summary 只兼容旧调用，不再展示。
    """
    _ = status, summary
    no = (work_order_no or "").strip()
    rows = deal_horizontal_rows(quote=quote, facts=facts, business_line=business_line)
    lines = [
        f"询价工单 {no}",
        DEAL_SUBTITLE,
        DEAL_HINT,
    ]
    lines.extend(f"{row['keyname']}：{row['value']}" for row in rows)
    lines.append(f"【{BTN_DEAL}】【{BTN_LOST}】【{BTN_NEGOTIATE}】")
    return "\n".join(lines)


def deal_wecom_payload(
    *,
    work_order_no: str,
    quote: dict[str, Any] | None = None,
    facts: dict[str, str] | None = None,
    card_version: int = 0,
    business_line: str = "",
) -> dict[str, Any]:
    """
    成交跟进卡：对齐现网 button_interaction。

    顶部来源红字；标题询价工单号；副标题跟进成交；三个按钮。
    字段：费用合计、报价有效期；海运/陆运再加时效。空运不写时效。
    """
    src = dict(quote or {})
    source = str(src.get("source_label") or "TMS标准报价").replace(" ", "")
    no = (work_order_no or "").strip()
    ver = int(card_version or 0)
    suffix = f":v{ver}" if ver > 0 else ""
    return {
        "msgtype": "template_card",
        "template_card": {
            "card_type": "button_interaction",
            "source": {"desc": source, "desc_color": 0},
            "main_title": {
                "title": f"询价工单 {no}",
                "desc": DEAL_SUBTITLE,
            },
            "sub_title_text": DEAL_HINT,
            "horizontal_content_list": deal_horizontal_rows(
                quote=src, facts=facts, business_line=business_line
            ),
            "task_id": card_task_id("deal", no),
            "button_list": [
                {
                    "text": BTN_DEAL,
                    "style": 1,
                    "key": f"{BTN_DEAL_KEY}:{no}{suffix}",
                },
                {
                    "text": BTN_LOST,
                    "style": 1,
                    "key": f"{BTN_LOST_KEY}:{no}{suffix}",
                },
                {
                    "text": BTN_NEGOTIATE,
                    "style": 1,
                    "key": f"{BTN_NEGOTIATE_KEY}:{no}{suffix}",
                },
            ],
        },
    }


def ticket_locked(*, work_order_no: str, status: str) -> str:
    """已成交/未成交/已关闭后，群里改字段或再报价。"""
    no = (work_order_no or "").strip() or "该工单"
    st = (status or "").strip() or "已关闭"
    return f"工单{no}{st}，不能修改工单信息"


def ticket_locked_file(*, work_order_no: str, status: str) -> str:
    """终态工单再发 Excel/PDF。"""
    no = (work_order_no or "").strip() or "该工单"
    st = (status or "").strip() or "已关闭"
    return f"工单{no}{st}，不能再上传附件"


def group_only_excel_pdf() -> str:
    return "群里只收 Excel 和 PDF，其它文件不会存入工单"


def group_only_product_can_quote() -> str:
    return "只有产品能报价格，请产品在群里发价格或 Excel/PDF"


_GROUP_QUOTE_TAIL = (
    "亦可通过“纯文字”告知新报价，或直接将新报价excel表发给智能体。\n"
    "（注：附件报价单仅作为文件存储于后台，智能体不读取且不将报价内容写进系统）"
)


def group_tms_confirm_hint(*, business_line: str = "") -> str:
    """
    有 TMS 价时写在报价有效期下方：谁可以确认沿用，以及怎么另报。

    空运喊航线人员，海运/陆运喊产品人员。附件只当存档，不读进报价。
    """
    line = (business_line or "").strip().upper()
    who = "航线人员" if line == "AIR" else "产品人员"
    return f"请{who}回复“确认报价”确定采用TMS报价。\n{_GROUP_QUOTE_TAIL}"


def group_no_tms_quote_hint(*, business_line: str = "") -> str:
    """
    TMS 无价时写在「暂无匹配」下方：谁可以用文字或 Excel 另报。

    空运喊航线人员，海运/陆运喊产品人员。附件只当存档，不读进报价。
    """
    line = (business_line or "").strip().upper()
    who = "航线人员" if line == "AIR" else "产品人员"
    return (
        f"{who}可通过“纯文字”告知新报价，或直接将新报价excel表发给智能体。\n"
        "（注：附件报价单仅作为文件存储于后台，智能体不读取且不将报价内容写进系统）"
    )


def quote_attachment_saved_app(work_order_no: str, count: int = 1) -> str:
    """群里告诉大家：模板或产品附件已进后台。"""
    no = (work_order_no or "").strip() or "WO"
    n = count if count > 0 else 1
    return f"工单{no}的{n}个附件已保存到后台工单"


def group_confirm_need_tms() -> str:
    return "本单还没有可沿用的 TMS 报价，请直接回复协同报价或上传报价附件。"


def group_ask_tms_or_product() -> str:
    return (
        "本单已有 TMS 报价。\n"
        "1. 要在 TMS 基础上改，请回复：在TMS基础上改（或直接回 1 / 第一个）\n"
        "2. 不要 TMS 报价、只用你刚说的价格，请回复：不用TMS报价（或直接回 2 / 第二个）"
    )


def group_tms_choice_unclear() -> str:
    return "还没确定怎么用 TMS 报价。请回复：在TMS基础上改 / 不用TMS报价，也可以直接回 1 或 2。"


def group_need_price_before_choice() -> str:
    return "请产品先发价格，再说在TMS基础上改或不用TMS报价"


def group_file_saved() -> str:
    return "附件已保存后台"


def group_quote_not_recognized() -> str:
    return "没有识别到报价，请产品再发一次价格或 Excel/PDF"


def group_fields_not_recognized(
    keys: tuple[str, ...] | None = None,
    business_line: str = "",
) -> str:
    """没认出协同字段。陆运列出本单可补项；海运列出协同五项（不含贸易条款）。"""
    if keys:
        line = (business_line or "").strip().upper() or "SEA"
        names = [collab_display_name(k, line) for k in keys]
        return "听到了。请直接说" + "、".join(names) + "。"
    return "听到了。请直接说货好时间、商品海关编码、是否含油/含电/含磁。"


def group_deal_stale(*, work_order_no: str) -> str:
    """点了上一版成交卡。"""
    no = (work_order_no or "").strip()
    return f"这张成交确认已过期。工单{no}已有更新的报价，请点击群里最新一条成交跟进。"


def group_deal_need_quote() -> str:
    return "还没有可确认的报价，请先完成报价再确认成交状态。"


def group_deal_sales_only() -> str:
    return "只有负责该工单的销售可以确认成交状态。"


def group_dissolved(*, work_order_no: str) -> str:
    no = (work_order_no or "").strip()
    return f"群已解散。工单{no}协同已结束。"


def group_cannot_recreate(*, work_order_no: str) -> str:
    """解散后再点拉群。"""
    no = (work_order_no or "").strip() or "该工单"
    return f"工单{no}协同已结束，不能再拉群。"


def group_deal_followup(*, sales_name: str, work_order_no: str) -> str:
    """
    群里文字成交跟进。只提示输入，不含报价合计、不含点选按钮。

    私聊成交卡仍走 deal_card，不改这里。
    """
    name = (sales_name or "").strip() or "销售"
    no = (work_order_no or "").strip()
    return (
        f"@{name}\n"
        f"询价工单 {no}\n"
        "报价单已发送，请确认成交状态\n"
        "咨询成交状态：请输入成交 / 未成交 (未成交需补充原因)/ 协商中"
    )


def group_deal_mention(*, sales_name: str, work_order_no: str) -> str:
    return group_deal_followup(sales_name=sales_name, work_order_no=work_order_no)


def deal_ack(
    outcome: str,
    work_order_no: str,
    next_followup_at: str = "",
    lost_reason: str = "",
) -> str:
    """成交三态回执。协商中写明顺延三个工作日和下次再问的月日 9：00。"""
    if outcome == "已成交":
        return f"已记录为「已成交」。工单 {work_order_no} 已关闭智能体跟进。"
    if outcome == "未成交":
        reason = (lost_reason or "").strip()
        if reason:
            return f"已记录为「未成交」。{lost_reason_ack(reason)}"
        return "已记录为「未成交」。请补充未成交原因。"
    when = _followup_month_day(next_followup_at)
    if when:
        return f"已记录为「协商中」。已顺延三个工作日，{when}的9：00再次发起成交状态询问。"
    return "已记录为「协商中」。已顺延三个工作日，第四个工作日早上9：00再次发起成交状态询问。"


def _followup_month_day(raw: str) -> str:
    """把主账再问时刻收成「9月24日」。"""
    from datetime import datetime

    text = (raw or "").strip()
    if not text:
        return ""
    for fmt, n in (("%Y-%m-%d %H:%M:%S", 19), ("%Y-%m-%d", 10)):
        try:
            dt = datetime.strptime(text[:n], fmt)
            return f"{dt.month}月{dt.day}日"
        except ValueError:
            continue
    return ""


def lost_reason_ack(reason: str = "") -> str:
    """收下未成交原因后回显原句，不再追问。"""
    text = (reason or "").strip()
    if text.endswith(("。", ".", "！", "!", "？", "?")):
        return f"已记下未成交原因：{text}"
    return f"已记下未成交原因：{text}。" if text else "已记下未成交原因。"


def deal_notify_card(
    *,
    work_order_no: str,
    sales_name: str,
    facts: dict[str, str] | None,
    quote: dict[str, Any] | None,
    business_line: str,
) -> str:
    """成交通知文字兜底：方案卡压缩。不写销售，避免标题出现「销售 销售」。"""
    no = (work_order_no or "").strip()
    _ = sales_name
    mode = scheme_transport_mode(business_line)
    rows = scheme_horizontal_rows(
        work_order_no=no,
        facts=dict(facts or {}),
        quote=dict(quote or {}),
        transport_mode=mode,
    )
    lines = [f"成交通知 工单 {no}"]
    lines.extend(f"{row['keyname']}：{row['value']}" for row in rows)
    return "\n".join(lines)


def site_card_url() -> str:
    """
    整张卡片点击后打开的后台地址。

    事件异常、系统异常、成交通知都走这里，点开是同一个网站。
    不用企业微信授权页，也不要用 work.weixin.qq.com。
    """
    from agent.config import get_settings

    base = str(getattr(get_settings(), "public_base_url", "") or "").strip().rstrip("/")
    return base or "https://ai.ytd-scm.com"


def deal_notify_wecom_payload(
    *,
    work_order_no: str,
    sales_name: str,
    facts: dict[str, str] | None,
    quote: dict[str, Any] | None,
    business_line: str,
) -> dict[str, Any]:
    """
    成交通知卡：走询价消息服务，不走询价小助手。

    字段按报价方案卡压缩，避免超过 6 格。标题不写销售。
    """
    no = (work_order_no or "").strip()
    _ = sales_name
    mode = scheme_transport_mode(business_line)
    rows = scheme_horizontal_rows(
        work_order_no=no,
        facts=dict(facts or {}),
        quote=dict(quote or {}),
        transport_mode=mode,
    )
    action_url = site_card_url()
    return {
        "msgtype": "template_card",
        "wecom_app": "notify",
        "template_card": {
            "card_type": "text_notice",
            "source": {"desc": "成交通知", "desc_color": 0},
            "main_title": {
                "title": f"成交通知 工单 {no}",
            },
            "horizontal_content_list": rows,
            "card_action": {"type": 1, "url": action_url},
        },
    }


def air_need_work_order() -> str:
    """空运群还没有当前单，且这句话里没有工单号。"""
    return "请先 @我 并发送要跟的工单号，例如 WO202609160001。"


def air_lock_need_work_order() -> str:
    """锁舱/释放必须自带工单号，不猜当前单。"""
    return "锁舱或释放舱位请带上工单号，例如 WO202609160001锁舱。"


def air_ticket_not_found(work_order_no: str) -> str:
    no = (work_order_no or "").strip()
    return f"找不到工单{no}，请核对工单号后再试。"


def air_ticket_not_air(work_order_no: str) -> str:
    no = (work_order_no or "").strip()
    return f"工单{no}不是空运单，不能在空运群里跟。"


def air_ticket_closed(work_order_no: str, status: str) -> str:
    no = (work_order_no or "").strip()
    st = (status or "").strip() or "已关闭"
    return f"工单{no}已是「{st}」，不能再激活。"


def air_ticket_occupied(work_order_no: str) -> str:
    no = (work_order_no or "").strip()
    return f"工单{no}已经在另一个群跟了，请到那个群继续。"


def air_group_brief(
    *,
    work_order_no: str,
    facts: dict[str, str],
    quote: dict[str, Any] | None,
    collab_facts: dict[str, str] | None,
    has_price: bool,
    tms_quote: dict[str, Any] | None = None,
    product_quote: dict[str, Any] | None = None,
    sales_name: str = "",
    airline_names: list[str] | None = None,
) -> str:
    """
    空运群激活摘要：与海运进群同一套结构，字段用空运。

    先 @ 销售和航线，再工单号 + 询价字段 + 协同补充（含电/含磁）+ TMS报价。
    TMS 段跟单聊同一套线路分项，总销售价只一行；禁止用「空运费=总价」冒充。
    不写「当前工单/询价要点/现在等到哪一步」。
    """
    no = (work_order_no or "").strip()
    at_bits: list[str] = []
    sales = (sales_name or "").strip()
    if sales:
        at_bits.append(f"@{sales}")
    for name in airline_names or []:
        token = f"@{(name or '').strip()}"
        if (name or "").strip() and token not in at_bits:
            at_bits.append(token)
    lines = ([(" ".join(at_bits))] if at_bits else []) + [f"工单号：{no}", "询价字段："]
    lines.extend(
        inquiry_field_lines(work_order_no=no, facts=facts, transport_mode="空运")[1:]
    )
    lines.append("")
    lines.append("协同补充字段：")
    collab = dict(collab_facts or {})
    from agent.schema.field_validate import air_general_cargo_collab_defaults

    # 品名是普货时，摘要里不能再把没写的含电/含磁显示成横杠。
    for key, val in air_general_cargo_collab_defaults(facts).items():
        current = (collab.get(key) or "").strip()
        if current in {"", "-", "—", "－"}:
            collab[key] = val
    for key in AIR_COLLAB_FIELDS:
        val = (collab.get(key) or "").strip()
        if val in {"", "-", "—", "－"}:
            val = "-"
        lines.append(f"{key}：{val}")
    tms, manual, adjust = _split_brief_quotes(
        quote=quote,
        tms_quote=tms_quote,
        product_quote=product_quote,
        has_price=has_price,
    )
    if tms:
        from agent.schema.air_route_options import project_group_air_quote

        tms = project_group_air_quote(tms, facts)
    lines.append("")
    lines.append("TMS报价：")
    if tms and (tms.get("total") or quote_fee_lines(tms)):
        lines.extend(quote_detail_lines(tms, business_line="AIR"))
        lines.append(group_tms_confirm_hint(business_line="AIR"))
    else:
        lines.append("TMS 暂无匹配报价")
        lines.append(group_no_tms_quote_hint(business_line="AIR"))
    if manual and (quote_fee_lines(manual) or str(manual.get("total") or "").strip() not in {"", "-"}):
        lines.append("")
        lines.append("人工报价：")
        lines.extend(_manual_quote_lines(manual, business_line="AIR"))
    if adjust and (quote_fee_lines(adjust) or str(adjust.get("total") or "").strip() not in {"", "-"}):
        lines.append("")
        lines.append("销售调价：")
        lines.extend(_manual_quote_lines(adjust, business_line="AIR"))
    return "\n".join(lines)


def air_activate_summary(
    *,
    work_order_no: str,
    facts: dict,
    status: str = "",
    wait_hint: str = "",
    quote: dict[str, Any] | None = None,
    collab_facts: dict[str, str] | None = None,
) -> str:
    """兼容旧名：激活成功改走海运同款群摘要。"""
    _ = status, wait_hint
    src = dict(quote or {})
    return air_group_brief(
        work_order_no=work_order_no,
        facts=dict(facts or {}),
        quote=src,
        collab_facts=collab_facts,
        has_price=bool(src.get("total") or src.get("quoteId")),
        tms_quote=src if quote_is_tms(src) else {},
        product_quote=src if quote_is_product(src) else {},
    )


def air_wait_hint(*, status: str, has_quote: bool, pending_lock: bool = False) -> str:
    """激活摘要里的「现在等到哪一步」。"""
    if pending_lock:
        return "等待航线选择要锁的舱位选项"
    st = (status or "").strip()
    if st == "协商中":
        return "协商中，可继续改价或确认成交"
    if has_quote or st == "已报价":
        return "已有报价，可补含电/含磁、航线改价、锁舱或确认成交"
    return "等待航线报价或锁舱（须先有 TMS 报价）"


def air_airline_only_quote() -> str:
    return "只有岗位为「航线·空运」的人能报价格。"


def air_airline_only_cabin() -> str:
    return "只有岗位为「航线·空运」的人能锁舱或释放舱位。"


def air_sales_only_deal() -> str:
    return "只有负责该工单的销售可以确认成交状态。"


def air_fields_not_recognized() -> str:
    return "听到了。空运群里可补充是否含电、是否含磁。"


def air_quote_not_recognized() -> str:
    return "没有识别到报价，请航线再发一次价格或 Excel/PDF。"


def air_fields_denied() -> str:
    return "只有这张单的销售或航线可以补含电/含磁。"


def air_no_tms_quote_to_lock(work_order_no: str) -> str:
    return air_lock_need_quote(work_order_no)


def air_lock_need_quote(work_order_no: str) -> str:
    """没有任何报价时的锁舱拒绝。不提 TMS，避免把人工报价说成没报价。"""
    no = (work_order_no or "").strip()
    return f"工单{no}还没有报价，不能锁舱。"


def air_ask_lock_option(options: list) -> str:
    lines = ["这张单有多条舱位选项，请回复要锁哪一条，例如 AIR-OPT-01："]
    for item in options:
        no = str(item.get("optionNo") or item.get("option_no") or "").strip()
        airline = str(item.get("airlineName") or item.get("airlineCode") or "").strip()
        if no and airline:
            lines.append(f"{no} {airline}")
        elif no:
            lines.append(no)
    return "\n".join(lines)


def air_lock_ok(
    *,
    work_order_no: str,
    operator: str,
    option_no: str,
    exec_no: str,
    quote_no: str,
) -> str:
    return (
        f"工单{work_order_no} 已通知TMS执行空运锁舱，执行成功。\n"
        f"操作人：{operator}\n"
        f"舱位选项：{option_no}\n"
        f"执行编号：{exec_no}\n"
        f"报价编号：{quote_no}"
    )


def air_lock_fail(*, work_order_no: str, operator: str, reason: str) -> str:
    why = (reason or "").strip() or "TMS未返回原因"
    return (
        f"工单{work_order_no} 空运锁舱失败。\n"
        f"操作人：{operator}\n"
        f"锁舱失败的原因：{why}\n"
        "已通知IT运维排查，请稍等。"
    )


def air_cabin_fail_reason(*, tms_reason: str = "", flight_date: str = "", today: str = "") -> str:
    """
    锁舱/释放失败原因：原样回 TMS 的 failReason。
    禁止改写成「报价选项不存在」或「航班已起飞」。
    flight_date / today 只为兼容旧调用，不再用来改写原文。
    """
    _ = flight_date
    _ = today
    return (tms_reason or "").strip()


def air_release_ok(
    *,
    work_order_no: str,
    operator: str,
    option_no: str,
    exec_no: str,
    quote_no: str,
) -> str:
    return (
        f"工单{work_order_no} 已通知TMS执行空运释放舱位成功\n"
        f"操作人：{operator}\n"
        f"舱位选项：{option_no}\n"
        f"执行编号：{exec_no}\n"
        f"报价编号：{quote_no}"
    )


def air_release_fail(*, work_order_no: str, operator: str, reason: str) -> str:
    why = (reason or "").strip() or "TMS未返回原因"
    return (
        f"工单{work_order_no} 空运释放舱位失败。\n"
        f"操作人：{operator}\n"
        f"释放舱位失败的原因：{why}\n"
        "已通知IT运维排查，请稍等。"
    )


def air_release_deal_blocked(work_order_no: str) -> str:
    no = (work_order_no or "").strip()
    return f"释放舱位失败，工单{no}已成交。"


def air_release_need_lock(work_order_no: str) -> str:
    no = (work_order_no or "").strip()
    return f"工单{no}还没有有效锁舱，不能释放。"


# —— 系统异常（UC024）：销售/群兜底与恢复；告警卡只给有权限的人 ——
TYPE_LLM = "大模型异常"
TYPE_TMS = "TMS异常"
TYPE_GROUP = "建群异常"

STEP_LLM_NO_TICKET = "llm_no_ticket"
STEP_LLM = "llm"
STEP_TMS_QUOTE = "tms_quote"
STEP_TMS_CABIN = "tms_cabin"
STEP_TMS_LOCK = "tms_lock"
STEP_TMS_RELEASE = "tms_release"
STEP_GROUP = "group"

SYS_EXC_LLM_FALLBACK = "大模型服务异常，IT排查中..."
SYS_EXC_TMS_FALLBACK = "TMS接口异常，IT运维排查中..."
SYS_EXC_RECOVERY_TMS = "TMS已恢复正常使用，感谢您的耐心等待。"
SYS_EXC_RECOVERY_LLM = "大模型已恢复正常使用，感谢您的耐心等待。"
SYS_EXC_RECOVERY_CABIN = "空运舱位接口已恢复正常使用，感谢您的耐心等待。"
SYS_EXC_RECOVERY_GROUP = "建群已恢复正常，感谢您的耐心等待。"

IMPACT_LLM_NO_TICKET = "尚未建单，不落后台工单"
IMPACT_LLM = "当前识别暂停，已有字段与工单状态保持不变"
IMPACT_TMS_QUOTE = "当前询价流程暂停，识别字段与原业务节点保持不变"
IMPACT_TMS_CABIN = "当前锁舱/释放暂停，工单状态保持不变"
IMPACT_TMS_LOCK = "锁舱失败，工单状态保持不变"
IMPACT_TMS_RELEASE = "释放舱位失败，工单状态保持不变"
IMPACT_GROUP = "当前建群/加人暂停，已建群不新建第二个群"


def system_exception_fallback(exception_type: str, *, names: list[str] | None = None) -> str:
    """判定系统异常后，给卡住的会话的兜底话术。"""
    kind = (exception_type or "").strip()
    if kind == TYPE_LLM:
        return SYS_EXC_LLM_FALLBACK
    if kind == TYPE_GROUP:
        people = [str(x).strip() for x in (names or []) if str(x).strip()]
        if not people:
            return "无法添加同事，请联系管理员检查企业微信配置。"
        joined = "、".join(people)
        return f"无法添加 {joined} 同事，请联系管理员检查企业微信配置。"
    return SYS_EXC_TMS_FALLBACK


def system_exception_recovery(step: str) -> str:
    """后台点重试成功后，先发这一句，再补没做完的事。"""
    key = (step or "").strip()
    if key == STEP_LLM or key == STEP_LLM_NO_TICKET:
        return SYS_EXC_RECOVERY_LLM
    if key in {STEP_TMS_CABIN, STEP_TMS_LOCK, STEP_TMS_RELEASE}:
        return SYS_EXC_RECOVERY_CABIN
    if key == STEP_GROUP:
        return SYS_EXC_RECOVERY_GROUP
    return SYS_EXC_RECOVERY_TMS


def system_exception_impact(step: str) -> str:
    """告警卡「影响范围」，按卡住的步骤。"""
    key = (step or "").strip()
    if key == STEP_LLM_NO_TICKET:
        return IMPACT_LLM_NO_TICKET
    if key == STEP_LLM:
        return IMPACT_LLM
    if key == STEP_TMS_LOCK:
        return IMPACT_TMS_LOCK
    if key == STEP_TMS_RELEASE:
        return IMPACT_TMS_RELEASE
    if key == STEP_TMS_CABIN:
        return IMPACT_TMS_CABIN
    if key == STEP_GROUP:
        return IMPACT_GROUP
    return IMPACT_TMS_QUOTE


def system_exception_alert_payload(
    *,
    work_order_no: str,
    exception_type: str,
    service: str,
    error_code: str,
    step: str,
    occurred_at: str,
    test_prefix: bool = False,
) -> dict[str, Any]:
    """
    系统异常通知卡：走询价消息服务，不走询价小助手。

    五行对齐现网示例：异常类型、接口/服务、错误码、影响范围、发生时间。
    标题不加 [test]：运维要看的就是这张卡本身，test_prefix 仅保留兼容入参。
    """
    no = (work_order_no or "").strip() or "无"
    _ = test_prefix
    title = "系统异常通知"
    rows = [
        {"keyname": "异常类型", "value": (exception_type or "").strip() or "-"},
        {"keyname": "接口/服务", "value": (service or "").strip() or "-"},
        {"keyname": "错误码", "value": (error_code or "").strip() or "-"},
        {"keyname": "影响范围", "value": system_exception_impact(step)},
        {"keyname": "发生时间", "value": (occurred_at or "").strip() or "-"},
    ]
    action_url = site_card_url()
    return {
        "msgtype": "template_card",
        "wecom_app": "notify",
        "template_card": {
            "card_type": "text_notice",
            "source": {"desc": "询价机器人系统告警", "desc_color": 0},
            "main_title": {
                "title": title,
                "desc": f"关联工单 {no}",
            },
            "horizontal_content_list": rows,
            "card_action": {"type": 1, "url": action_url},
        },
    }


def system_exception_alert_text(
    *,
    work_order_no: str,
    exception_type: str,
    service: str,
    error_code: str,
    step: str,
    occurred_at: str,
    test_prefix: bool = False,
) -> str:
    """告警卡纯文本兜底（通道降级时仍能看懂）。标题同样不加 [test]。"""
    no = (work_order_no or "").strip() or "无"
    _ = test_prefix
    head = "系统异常通知"
    lines = [
        "询价机器人系统告警",
        head,
        f"关联工单 {no}",
        f"异常类型：{(exception_type or '').strip() or '-'}",
        f"接口/服务：{(service or '').strip() or '-'}",
        f"错误码：{(error_code or '').strip() or '-'}",
        f"影响范围：{system_exception_impact(step)}",
        f"发生时间：{(occurred_at or '').strip() or '-'}",
    ]
    return "\n".join(lines)


def event_exception_session_text(
    *,
    exception_type: str,
    work_order_no: str,
    names: list[str] | None,
    hours: float,
    mention: bool,
    test_prefix: bool = False,
) -> str:
    """
    事件异常会话两行话术。

    mention=True 时姓名前加 @，多人空格拼接。小时数用后台规则值。
    测服也不加 [test]：这是销售/航线/产品会话，不是运维告警。
    """
    from agent.policy.event_exception import format_timeout_hours

    _ = test_prefix
    no = (work_order_no or "").strip() or "-"
    kind = (exception_type or "").strip() or "事件异常"
    hour_text = format_timeout_hours(hours)
    people = [str(n).strip() for n in (names or []) if str(n).strip()]
    if mention:
        tagged = " ".join(f"@{n}" for n in people) if people else "相关人员"
    else:
        tagged = " ".join(people) if people else "您"
    body = (
        f"关联询价工单 {no}。{tagged}已超过 {hour_text} 工作小时 无响应，"
        f"事件异常已记录工单，请及时处理。"
    )
    return f"{kind}\n{body}"


def event_exception_alert_payload(
    *,
    card_title: str,
    work_order_no: str,
    hours: float,
    test_prefix: bool = False,
) -> dict[str, Any]:
    """
    事件异常通知卡：走询价消息服务，不走询价小助手。

    对齐现网「询价机器人事件告警」：标题用响应异常，正文用「您已超过」。
    测服标题也不加 [test]；test_prefix 仅保留兼容入参。
    """
    from agent.policy.event_exception import format_timeout_hours

    _ = test_prefix
    no = (work_order_no or "").strip() or "-"
    title = (card_title or "").strip() or "事件响应异常"
    hour_text = format_timeout_hours(hours)
    action_url = site_card_url()
    return {
        "msgtype": "template_card",
        "wecom_app": "notify",
        "template_card": {
            "card_type": "text_notice",
            "source": {"desc": "询价机器人事件告警", "desc_color": 0},
            "main_title": {
                "title": title,
                "desc": f"关联询价工单 {no}",
            },
            "sub_title_text": f"您已超过 {hour_text} 工作小时无响应，事件异常已记录工单，请及时处理。",
            "card_action": {"type": 1, "url": action_url},
        },
    }


def event_exception_alert_text(
    *,
    card_title: str,
    work_order_no: str,
    hours: float,
    test_prefix: bool = False,
) -> str:
    """应用卡片纯文本兜底。"""
    payload = event_exception_alert_payload(
        card_title=card_title,
        work_order_no=work_order_no,
        hours=hours,
        test_prefix=test_prefix,
    )
    card = payload.get("template_card") or {}
    main = card.get("main_title") or {}
    lines = [
        str((card.get("source") or {}).get("desc") or "询价机器人事件告警"),
        str(main.get("title") or ""),
        str(main.get("desc") or ""),
        str(card.get("sub_title_text") or ""),
    ]
    return "\n".join(x for x in lines if x)