qizhong/inquiry_robot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
ab6923d8cceb0f39305580d6377987e017527eed
inquiry_robot/inquiry-agent/tests/test_sea_text_inquiry.py
T

2561 lines
103 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
海运文字询价主路径单测（不连企微 / 真实 TMS / 8180）。
覆盖：海运补问、询价确认卡、有价/无价卡、跳过协同、拉群、群里静默补协同字段。
空运话术不得出现在海运路径。
"""
from __future__ import annotations
import os
import sys
import unittest
ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
if ROOT not in sys.path:
sys.path.insert(0, ROOT)
os.environ.setdefault("YTD_ENV", "test")
os.environ.setdefault("REDIS_BACKEND", "memory")
os.environ.setdefault("CHECKPOINT_BACKEND", "memory")
os.environ.setdefault("MESSAGE_STORE_BACKEND", "memory")
os.environ.setdefault("LLM_DATA_USAGE_CONFIRMED", "false")
os.environ.setdefault("LLM_ALLOW_NETWORK", "false")
os.environ.setdefault("LEDGER_BACKEND", "memory")
from agent.channel.wecom.models import InboundMessage
from agent.handlers.group_collab import handle_group_collab
from agent.handlers.text_inquiry import handle_text_inquiry
from agent.ledger.memory_ledger import MemoryLedger
from agent.policy import inquiry_copy as copy
from agent.policy.air_text_flow import FlowSession
from agent.policy.sea_text_flow import SeaTextInquiryFlow, reset_sea_text_flow_for_test
from agent.schema.field_validate import SEA_COLLAB_FIELDS, validate_required_fields
from agent.schema.tms_sea_query import assemble_sea_query
COMPLETE_SEA = {
"起运港": "上海",
"目的港": "洛杉矶",
"品名": "普货",
"货量": "20吨",
"整柜或拼柜": "整柜",
"箱型箱量": "1x40HQ",
"贸易条款": "FOB",
"运输分类": "Port to Port",
}
def drive_sea(flow, **kwargs):
"""字段齐时先出核对卡；要看到查价结果的旧用例再自动回一句确定。"""
phase = flow.on_text(**kwargs)
if phase == "wait_confirm":
phase = flow.on_text(sender_id=kwargs["sender_id"], text="确定", reply=kwargs["reply"])
return phase
class SeaCollabContractTests(unittest.TestCase):
def test_collab_no_longer_asks_trade_terms(self):
self.assertNotIn("贸易条款", SEA_COLLAB_FIELDS)
self.assertEqual(
SEA_COLLAB_FIELDS,
("货好时间", "HS编码", "是否含油", "是否含电", "是否含磁"),
)
class FakeGroupClient:
"""单测假企微建群：不打外网。"""
def __init__(self) -> None:
self.created: list[dict] = []
self.sent: list[dict] = []
self.fail = False
self.next_chat_id = "wr_sea_1"
def create_group(self, *, name: str, userids: list[str]) -> dict:
if self.fail:
return {"ok": False, "error": "wecom_create_fail", "chat_id": ""}
self.created.append({"name": name, "userids": list(userids)})
return {"ok": True, "chat_id": self.next_chat_id, "error": ""}
def add_group_members(self, *, chat_id: str, userids: list[str]) -> dict:
self.created.append({"name": "", "userids": list(userids), "chat_id": chat_id, "add": True})
return {"ok": True}
def send_group(self, *, chat_id: str, content: str, mention_userids: list | None = None) -> dict:
self.sent.append(
{
"chat_id": chat_id,
"content": content,
"kind": "text",
"mention_userids": list(mention_userids or []),
}
)
return {"ok": True}
def send_group_file(self, *, chat_id: str, filename: str, filepath: str = "", file_b64: str = "") -> dict:
self.sent.append(
{"chat_id": chat_id, "filename": filename, "kind": "file", "file_b64": file_b64}
)
return {"ok": True}
def send_group_card(self, *, chat_id: str, template_card: dict) -> dict:
self.sent.append({"chat_id": chat_id, "kind": "card", "template_card": template_card})
return {"ok": True}
def send_text(self, *, touser: str, content: str) -> dict:
self.sent.append({"touser": touser, "content": content, "kind": "app_text"})
return {"ok": True}
def dismiss_group(self, *, chat_id: str, userids: list) -> dict:
self.sent.append({"chat_id": chat_id, "kind": "dismiss", "userids": list(userids)})
return {"ok": True}
class SeaTextInquiryTests(unittest.TestCase):
def setUp(self) -> None:
self.ledger = MemoryLedger()
self.ledger.set_quote_templates(
[
{
"templateId": "mem-default-sea",
"name": "海运默认",
"bizType": "海运",
"keywords": [],
"isDefault": True,
}
]
)
self.groups = FakeGroupClient()
self.flow = SeaTextInquiryFlow(self.ledger, group_client=self.groups)
self.replies: list[str] = []
self.extras: list[dict] = []
def _reply(self, text: str, extra=None) -> None:
self.replies.append(text)
self.extras.append(dict(extra) if extra else {})
def test_complete_fields_hold_confirm_before_tms(self) -> None:
"""字段齐了只出核对卡，不建单、不查价。"""
phase = self.flow.on_text(
sender_id="s-hold",
text="海运完整",
reply=self._reply,
injected_facts=COMPLETE_SEA,
injected_mode="SEA",
)
self.assertEqual(phase, "wait_confirm")
card = self.replies[-1]
self.assertIn(copy.SEA_CONFIRM_TAIL, card)
self.assertIn("运输方式（必填）：海运", card)
self.assertIn("货物数量（必填）：20吨", card)
self.assertIn("贸易条款（必填）：FOB", card)
self.assertNotIn("工单号", card)
self.assertNotIn("正在查询TMS报价", card)
sess = self.flow.session_of("s-hold")
self.assertEqual(sess.phase, "wait_confirm")
self.assertFalse((sess.work_order_no or "").strip())
self.assertEqual(self.ledger._tickets, {})
def test_confirm_phrases_then_query_tms(self) -> None:
"""确定、确认、查价、可以查了 都建单并查价。"""
for word in ("确定", "确认", "查价", "可以查了"):
with self.subTest(word=word):
self.replies.clear()
sender = f"s-go-{word}"
held = self.flow.on_text(
sender_id=sender,
text="海运完整",
reply=self._reply,
injected_facts=COMPLETE_SEA,
injected_mode="SEA",
)
self.assertEqual(held, "wait_confirm")
self.replies.clear()
phase = self.flow.on_text(sender_id=sender, text=word, reply=self._reply)
self.assertEqual(phase, "wait_collab")
joined = "\n".join(self.replies)
self.assertIn("正在查询TMS报价", joined)
self.assertIn("工单号：", joined)
sess = self.flow.session_of(sender)
self.assertTrue((sess.work_order_no or "").startswith("WO"))
def test_resend_while_waiting_merges_and_does_not_quote(self) -> None:
"""核对卡出来后改一处，并进原字段再出卡，仍不建单。"""
phase = self.flow.on_text(
sender_id="s-fix",
text="海运完整",
reply=self._reply,
injected_facts=COMPLETE_SEA,
injected_mode="SEA",
)
self.assertEqual(phase, "wait_confirm")
self.replies.clear()
fixed = dict(COMPLETE_SEA)
fixed["目的港"] = "纽约"
phase = self.flow.on_text(
sender_id="s-fix",
text="目的港改成纽约",
reply=self._reply,
injected_facts=fixed,
injected_mode="SEA",
)
self.assertEqual(phase, "wait_confirm")
self.assertIn("目的港（必填）：纽约", self.replies[-1])
self.assertIn(copy.SEA_CONFIRM_TAIL, self.replies[-1])
sess = self.flow.session_of("s-fix")
self.assertEqual(sess.facts.get("目的港"), "纽约")
self.assertEqual(sess.facts.get("起运港"), "上海")
self.assertFalse((sess.work_order_no or "").strip())
self.assertEqual(self.ledger._tickets, {})
def test_sea_pieces_fill_cargo_qty(self) -> None:
facts = {
"起运港": "上海",
"目的港": "洛杉矶",
"品名": "普货",
"件数": "15件",
"整柜或拼柜": "整柜",
"箱型箱量": "1x40HQ",
"贸易条款": "FOB",
"运输分类": "Port to Port",
}
check = validate_required_fields(facts=facts, business_line="SEA")
self.assertTrue(check["ok"])
self.assertEqual(check["facts"]["货量"], "15件")
self.assertNotIn("货量", check["missing"])
def test_container_shaped_cargo_qty_uses_pieces(self) -> None:
"""货量被抽成箱型时，改用件数。20件、15件同一条规则。"""
facts = {
"起运港": "广州港",
"目的港": "巴生港",
"品名": "机器设备",
"货量": "40HQ*1",
"件数": "20件",
"整柜或拼柜": "整柜",
"箱型箱量": "40HQ*1",
"贸易条款": "DOOR-DOOR",
"运输分类": "Port to Port",
}
check = validate_required_fields(facts=facts, business_line="SEA")
self.assertEqual(check["facts"]["货量"], "20件")
self.assertEqual(check["facts"]["箱型箱量"], "40HQ*1")
def test_sea_missing_container_only_asks_container(self) -> None:
facts = {
"起运港": "上海",
"目的港": "洛杉矶",
"品名": "茶叶",
"货量": "20吨",
"整柜或拼柜": "整柜",
"贸易条款": "FOB",
"运输分类": "Port to Port",
}
check = validate_required_fields(facts=facts, business_line="SEA")
self.assertFalse(check["ok"])
self.assertEqual(check["missing"], ["箱型箱量"])
self.assertIn("箱型箱量", check["missing_display"])
self.assertNotIn("件数", check["missing"])
self.assertNotIn("毛重", check["missing"])
text = copy.ask_clarify(
facts=check["facts"],
missing_keys=check["missing"],
transport_mode="海运",
)
self.assertIn("运输方式：海运", text)
self.assertIn("起运港：上海", text)
self.assertIn("货物数量：20吨", text)
self.assertIn("1.箱型箱量（必填）：\n(如：1x40HQ）", text)
self.assertNotIn("箱型箱量：(如", text)
self.assertNotIn("件数", text)
self.assertNotIn("重量", text)
self.assertNotIn("报价日期：(如", text)
self.assertIn("报价日期：", text)
def test_sea_requires_trade_terms_and_class(self) -> None:
facts = {
"起运港": "上海",
"目的港": "洛杉矶",
"品名": "普货",
"货量": "20吨",
"整柜或拼柜": "整柜",
"箱型箱量": "1x40HQ",
}
check = validate_required_fields(facts=facts, business_line="SEA")
self.assertFalse(check["ok"])
self.assertIn("贸易条款", check["missing"])
self.assertIn("运输分类", check["missing"])
self.assertIn("贸易条款", check["missing_display"])
self.assertIn("运输分类", check["missing_display"])
text = copy.ask_clarify(
facts=check["facts"],
missing_keys=check["missing"],
transport_mode="海运",
)
self.assertIn("贸易条款（必填）：\n(如：门到门、港到港）", text)
self.assertNotIn("如：FOB", text)
def test_sea_rejects_door_to_door_class(self) -> None:
facts = {
"起运港": "上海",
"目的港": "洛杉矶",
"品名": "普货",
"货量": "20吨",
"整柜或拼柜": "整柜",
"箱型箱量": "1x40HQ",
"贸易条款": "FOB",
"运输分类": "门到门",
}
check = validate_required_fields(facts=facts, business_line="SEA")
self.assertFalse(check["ok"])
self.assertIn("运输分类", check["missing"])
def test_sea_canonicalizes_p2p(self) -> None:
facts = {
"起运港": "上海",
"目的港": "洛杉矶",
"品名": "普货",
"货量": "20吨",
"整柜或拼柜": "整柜",
"箱型箱量": "1x40HQ",
"贸易条款": "到岸",
"运输分类": "p2p",
}
check = validate_required_fields(facts=facts, business_line="SEA")
self.assertTrue(check["ok"])
self.assertEqual(check["facts"]["运输分类"], "Port to Port")
def test_inquiry_card_lists_ten_fields(self) -> None:
lines = copy.inquiry_field_lines(
work_order_no="WO202609200001",
facts={
"起运港": "上海",
"目的港": "洛杉矶",
"品名": "普货",
"货量": "20吨",
"整柜或拼柜": "整柜",
"箱型箱量": "1x40HQ",
"贸易条款": "FOB",
"运输分类": "Port to Port",
"报价日期": "2026-09-20",
},
transport_mode="海运",
)
text = "\n".join(lines)
self.assertIn("工单号：WO202609200001", text)
self.assertIn("贸易条款：FOB", text)
self.assertIn("运输分类：Port to Port", text)
self.assertLess(text.index("箱型箱量"), text.index("贸易条款"))
self.assertLess(text.index("贸易条款"), text.index("运输分类"))
def test_clarify_asks_class_as_two_options(self) -> None:
text = copy.ask_clarify(
facts={
"起运港": "上海",
"目的港": "洛杉矶",
"品名": "服装",
"货量": "1柜",
"整柜或拼柜": "整柜",
"箱型箱量": "1x40HQ",
"贸易条款": "FOB",
},
missing_keys=["运输分类"],
transport_mode="海运",
)
self.assertIn("1.运输分类（必填）：\n(如：Port to Port、其他）", text)
self.assertIn("贸易条款：FOB", text)
def test_incomplete_without_trade_terms_does_not_create_ticket(self) -> None:
facts = dict(COMPLETE_SEA)
facts.pop("贸易条款")
phase = drive_sea(self.flow,
sender_id="s-miss",
text="海运缺条款",
reply=self._reply,
injected_facts=facts,
injected_mode="SEA",
)
self.assertEqual(phase, "clarify")
self.assertIn("贸易条款", "\n".join(self.replies))
self.assertFalse((self.flow.session_of("s-miss").work_order_no or "").startswith("WO"))
def test_leftover_door_to_door_not_written_as_trade_terms(self) -> None:
"""旧草稿里的门到门不能写到确认卡贸易条款。测服 WO202609200013。"""
leftover = {
"起运港": "纽约港",
"目的港": "巴生西港",
"品名": "机器设备",
"货量": "15件",
"整柜或拼柜": "整柜",
"箱型箱量": "40HQ*1",
"贸易条款": "门到门",
}
self.flow._save(
FlowSession(
sender_id="s-door",
thread_id="s-door",
phase="clarify",
business_line="SEA",
facts=leftover,
)
)
phase = drive_sea(self.flow,
sender_id="s-door",
text="海运、纽约港到巴生西港，机器设备，40HQ*1，整柜，1.5KG，15CBM，15件",
reply=self._reply,
injected_facts=dict(leftover),
injected_mode="SEA",
)
self.assertEqual(phase, "clarify")
joined = "\n".join(self.replies)
self.assertNotIn("贸易条款：门到门", joined)
self.assertIn("贸易条款", joined)
self.assertNotEqual(self.flow.session_of("s-door").facts.get("贸易条款"), "门到门")
def test_reply_door_to_door_fills_trade_terms(self) -> None:
"""补问后只回门到门，要记成贸易条款，不能回参数提取不全。"""
first = dict(COMPLETE_SEA)
first.pop("贸易条款")
phase = drive_sea(self.flow,
sender_id="s-door-reply",
text="先缺条款",
reply=self._reply,
injected_facts=first,
injected_mode="SEA",
)
self.assertEqual(phase, "clarify")
self.replies.clear()
phase = drive_sea(self.flow,
sender_id="s-door-reply",
text="门到门",
reply=self._reply,
)
self.assertEqual(phase, "wait_collab")
joined = "\n".join(self.replies)
self.assertIn("贸易条款：门到门", joined)
self.assertNotIn("参数提取不全", joined)
def test_door_in_sentence_kept_after_choosing_sea(self) -> None:
"""没写海运、句中有门到门，回「2」后贸易条款要带上，不能再追问。"""
from agent.channel.queue import MemoryMessageStore
from agent.policy.air_text_flow import reset_air_text_flow_for_test
from agent.policy.land_text_flow import reset_land_text_flow_for_test
from agent.policy.sea_text_flow import get_sea_text_flow
reset_sea_text_flow_for_test()
reset_land_text_flow_for_test()
reset_air_text_flow_for_test()
store = MemoryMessageStore()
text = "广州港到巴生港，机器设备，40HQ*1,整柜，1.5KG,15CBM,15件，Port to Port,门到门"
facts = {
"起运港": "广州港",
"目的港": "巴生港",
"品名": "机器设备",
"货量": "15件",
"整柜或拼柜": "整柜",
"箱型箱量": "40HQ*1",
"运输分类": "Port to Port",
}
phase = handle_text_inquiry(
InboundMessage(sender_id="s-door-mode", message_id="m-door-1", content=text),
injected_facts=facts,
store=store,
)
self.assertEqual(phase, "need_mode")
phase = handle_text_inquiry(
InboundMessage(sender_id="s-door-mode", message_id="m-door-2", content="2"),
store=store,
)
body = "\n".join(item.content for item in store._outbox.values())
sess = get_sea_text_flow().session_of("s-door-mode")
self.assertEqual(sess.facts.get("贸易条款"), "门到门")
self.assertEqual(sess.facts.get("运输分类"), "Port to Port")
self.assertIn("贸易条款（必填）：门到门", body)
self.assertIn(copy.SEA_CONFIRM_TAIL, body)
self.assertEqual(phase, "wait_confirm")
self.assertNotIn("1.贸易条款（必填）", body)
self.assertNotEqual(phase, "need_mode")
def test_port_to_port_phrase_kept_after_choosing_sea(self) -> None:
"""没写海运、句中有港到港，回「2」后贸易条款是港到港，运输分类仍是 Port to Port。"""
from agent.channel.queue import MemoryMessageStore
from agent.policy.air_text_flow import reset_air_text_flow_for_test
from agent.policy.land_text_flow import reset_land_text_flow_for_test
from agent.policy.sea_text_flow import get_sea_text_flow
reset_sea_text_flow_for_test()
reset_land_text_flow_for_test()
reset_air_text_flow_for_test()
store = MemoryMessageStore()
text = "广州港到巴生港，机器设备，40HQ*1,整柜，1.5KG,15CBM,15件，Port to Port，港到港"
facts = {
"起运港": "广州港",
"目的港": "巴生港",
"品名": "机器设备",
"货量": "15件",
"整柜或拼柜": "整柜",
"箱型箱量": "40HQ*1",
"运输分类": "Port to Port",
}
phase = handle_text_inquiry(
InboundMessage(sender_id="s-port-mode", message_id="m-port-1", content=text),
injected_facts=facts,
store=store,
)
self.assertEqual(phase, "need_mode")
phase = handle_text_inquiry(
InboundMessage(sender_id="s-port-mode", message_id="m-port-2", content="2"),
store=store,
)
body = "\n".join(item.content for item in store._outbox.values())
sess = get_sea_text_flow().session_of("s-port-mode")
self.assertEqual(sess.facts.get("贸易条款"), "港到港")
self.assertEqual(sess.facts.get("运输分类"), "Port to Port")
self.assertIn("贸易条款（必填）：港到港", body)
self.assertIn(copy.SEA_CONFIRM_TAIL, body)
self.assertEqual(phase, "wait_confirm")
self.assertNotIn("1.贸易条款（必填）", body)
self.assertNotEqual(phase, "need_mode")
def test_harvest_class_from_reply_then_quotes(self) -> None:
first = dict(COMPLETE_SEA)
first.pop("运输分类")
phase = drive_sea(self.flow,
sender_id="s-class",
text="先缺分类",
reply=self._reply,
injected_facts=first,
injected_mode="SEA",
)
self.assertEqual(phase, "clarify")
self.replies.clear()
phase = drive_sea(self.flow,
sender_id="s-class",
text="Port to Port",
reply=self._reply,
injected_mode="SEA",
)
self.assertEqual(phase, "wait_collab")
joined = "\n".join(self.replies)
self.assertIn("运输分类：Port to Port", joined)
self.assertIn("贸易条款：FOB", joined)
def test_complete_sea_inquiry_card_uses_ten_fields(self) -> None:
phase = drive_sea(self.flow,
sender_id="s1",
text="海运完整",
reply=self._reply,
injected_facts=COMPLETE_SEA,
injected_mode="SEA",
)
self.assertEqual(phase, "wait_collab")
joined = "\n".join(self.replies)
self.assertIn("运输方式：海运", joined)
self.assertIn("起运港：上海", joined)
self.assertIn("目的港：洛杉矶", joined)
self.assertIn("货物数量：20吨", joined)
self.assertIn("整柜或拼柜：整柜", joined)
self.assertIn("箱型箱量：1x40HQ", joined)
self.assertIn("贸易条款：FOB", joined)
self.assertIn("运输分类：Port to Port", joined)
self.assertIn("正在查询TMS报价", joined)
self.assertNotIn("件数：", joined.split("正在查询TMS报价")[0])
self.assertNotIn("本轮先开通空运", joined)
sess = self.flow.session_of("s1")
self.assertEqual(sess.business_line, "SEA")
self.assertEqual(sess.status, "已报价")
def test_new_round_after_ticket_does_not_reuse_old_facts(self) -> None:
"""上一单已出号，新一轮不能带上旧贸易条款。"""
first = drive_sea(self.flow,
sender_id="s-round",
text="海运第一轮",
reply=self._reply,
injected_facts=COMPLETE_SEA,
injected_mode="SEA",
)
self.assertEqual(first, "wait_collab")
old_no = self.flow.session_of("s-round").work_order_no
self.assertTrue((old_no or "").startswith("WO"))
self.replies.clear()
incomplete = {
"起运港": "纽约港",
"目的港": "巴生西港",
"品名": "机器设备",
"货量": "15件",
"整柜或拼柜": "整柜",
"箱型箱量": "40HQ*1",
"运输分类": "Port to Port",
}
phase = drive_sea(self.flow,
sender_id="s-round",
text="海运、纽约港到巴生西港，机器设备，40HQ*1，整柜，1.5KG，15CBM，15件，Port to Port",
reply=self._reply,
injected_facts=incomplete,
injected_mode="SEA",
)
self.assertEqual(phase, "clarify")
joined = "\n".join(self.replies)
self.assertNotIn("贸易条款：FOB", joined)
self.assertNotIn("贸易条款：门到门", joined)
self.assertIn("贸易条款", joined)
sess = self.flow.session_of("s-round")
self.assertNotEqual(sess.facts.get("贸易条款"), "FOB")
self.assertNotEqual(sess.work_order_no, old_no)
def test_handler_routes_sea_not_air_deferred(self) -> None:
reset_sea_text_flow_for_test()
from agent.channel.queue import MemoryMessageStore
store = MemoryMessageStore()
msg = InboundMessage(
sender_id="s2",
message_id="m-sea-1",
content="海运询价",
)
phase = handle_text_inquiry(
msg,
injected_facts=COMPLETE_SEA,
injected_mode="SEA",
flow=self.flow,
store=store,
)
self.assertEqual(phase, "wait_confirm")
self.assertNotEqual(phase, "sea_deferred")
class SeaTmsCardTests(unittest.TestCase):
def setUp(self) -> None:
self.ledger = MemoryLedger()
self.groups = FakeGroupClient()
self.flow = SeaTextInquiryFlow(self.ledger, group_client=self.groups)
self.replies: list[str] = []
self.extras: list[dict] = []
def _reply(self, text: str, extra=None) -> None:
self.replies.append(text)
self.extras.append(dict(extra) if extra else {})
def test_assemble_sea_query_does_not_need_air_measures(self) -> None:
ready = assemble_sea_query({**COMPLETE_SEA, "贸易条款": "FOB", "运输分类": "Port to Port"})
self.assertTrue(ready["ok"])
self.assertEqual(ready["missing"], [])
self.assertNotIn("件数", ready.get("missing") or [])
def test_assemble_requires_trade_terms_not_class(self) -> None:
src = {
"起运港": "上海",
"目的港": "洛杉矶",
"品名": "普货",
"货量": "20吨",
"整柜或拼柜": "整柜",
"箱型箱量": "1x40HQ",
"运输分类": "Port to Port",
}
missing = assemble_sea_query(src)
self.assertFalse(missing["ok"])
self.assertIn("贸易条款", missing["missing"])
self.assertNotIn("运输分类", missing["missing"])
src["贸易条款"] = "FOB"
ready = assemble_sea_query(src)
self.assertTrue(ready["ok"])
self.assertEqual(ready["tms_facts"]["贸易条款"], "FOB")
self.assertNotIn("运输分类", ready["tms_facts"])
self.assertEqual(ready["facts"]["运输分类"], "Port to Port")
def test_sea_quote_text_uses_three_buckets(self) -> None:
quote = {
"total": "6952.77",
"currency": "CNY",
"source_label": "TMS 标准报价",
"polCostItems": [
{"name": "操作费", "amount": "1020", "currency": "CNY"},
{"name": "舱单", "amount": "50", "currency": "CNY"},
],
"oceanCostItems": [{"name": "海运费", "amount": "575", "currency": "USD"}],
"podCostItems": [
{"name": "文件费", "amount": "450", "currency": "CNY"},
{"name": "设备费", "amount": "30", "currency": "CNY"},
],
"fee_total_polCostItems": "1070",
"fee_total_oceanCostItems": "4038",
"fee_total_podCostItems": "470",
}
text = copy.sea_quote_detail_text(work_order_no="WO202609200002", quote=quote)
self.assertIn("始发港费用\n操作费：CNY 1020", text)
self.assertIn("始发港费用小计：CNY 1070", text)
self.assertIn("海运费：USD 575", text)
self.assertIn("海运费小计：CNY 4038", text)
self.assertIn("目的港费用小计：CNY 470", text)
self.assertIn("费用合计：CNY 6952.77", text)
self.assertNotIn("时效", text)
def test_long_fee_lines_go_to_text_not_clipped_card(self) -> None:
"""海运多项费用：完整明细必须在文本里，卡片引用条不得半截。"""
long_quote = {
"total": "2287.65",
"currency": "CNY",
"eta": "1天",
"valid_until": "2026-09-30",
"source_label": "TMS 标准报价",
"fee_lines": [
{"name": "海运费", "amount": "100.00", "currency": "USD"},
{"name": "燃油附加费", "amount": "15.00", "currency": "USD"},
{"name": "操作费", "amount": "680.00", "currency": "CNY"},
{"name": "文件费", "amount": "450.00", "currency": "CNY"},
{"name": "封条费（SEAI）", "amount": "50.00", "currency": "CNY"},
{"name": "设备费", "amount": "50.00", "currency": "CNY"},
{"name": "报关费", "amount": "250.00", "currency": "CNY"},
],
}
text = copy.sea_quote_detail_text(
work_order_no="WO202609150030", quote=long_quote
)
self.assertEqual(
text,
"\n".join(
[
"TMS标准报价",
"WO202609150030价格明细",
"海运费：USD 100.00",
"燃油附加费：USD 15.00",
"操作费：CNY 680.00",
"文件费：CNY 450.00",
"封条费（SEAI）：CNY 50.00",
"设备费：CNY 50.00",
"报关费：CNY 250.00",
"费用合计：CNY 2287.65",
]
),
)
self.assertNotIn("时效", text)
self.assertNotIn("报价有效期", text)
card_text = copy.quote_area_safe_text(long_quote, business_line="SEA")
self.assertEqual(
card_text,
"费用合计：CNY 2287.65\n共7项费用，详见上一条明细",
)
short_quote = {
"total": "100",
"currency": "CNY",
"fee_lines": [{"name": "海运费", "amount": "100", "currency": "CNY"}],
}
self.assertEqual(
copy.quote_area_safe_text(short_quote, business_line="SEA"),
"费用合计：CNY 100\n共1项费用，详见上一条明细",
)
payload = copy.sea_tms_hit_wecom_payload(
work_order_no="WO202609150030", quote=long_quote
)
self.assertEqual(
payload["template_card"]["quote_area"]["quote_text"], card_text
)
sub = payload["template_card"]["sub_title_text"]
self.assertIn("时效：1天", sub)
self.assertIn("报价有效期：2026-09-30", sub)
def test_sea_hit_sends_detail_text_before_card(self) -> None:
"""
有价必须先发出价格明细文字，再发卡。
测服 WO202609200020：卡上写「详见上一条明细」，销售却先看到卡。
"""
steps: list[str] = []
def reply(text: str, extra=None) -> str:
kind = "card" if extra else "text"
steps.append(kind)
return f"out-{len(steps)}"
def wait(out_id: str, *, work_order_no: str, what: str) -> None:
steps.append(f"wait:{out_id}")
self.flow._wait_outbound_sent = wait # type: ignore[method-assign]
self.flow._detail_before_card_gap_sec = 0.0
self.flow._emit_sea_hit(
reply,
work_order_no="WO202609200020",
quote={
"total": "14298.53",
"currency": "CNY",
"source_label": "TMS标准报价",
"fee_lines": [{"name": "操作费", "amount": "1220", "currency": "CNY"}],
},
)
self.assertEqual(steps[:3], ["text", "wait:out-1", "card"])
self.assertIn("card", steps)
def test_sea_hit_card_has_price_and_two_buttons(self) -> None:
phase = drive_sea(self.flow,
sender_id="s1",
text="海运有价",
reply=self._reply,
injected_facts=COMPLETE_SEA,
injected_mode="SEA",
)
self.assertEqual(phase, "wait_collab")
joined = "\n".join(self.replies)
self.assertIn("价格明细", joined)
self.assertIn("是否拉产品人员进群协同", joined)
self.assertIn(copy.BTN_PULL_COLLAB, joined)
self.assertIn(copy.BTN_SKIP_COLLAB_TMS, joined)
self.assertNotIn(copy.BTN_SKIP_COLLAB, joined)
self.assertEqual(copy.canonical_button(copy.BTN_SKIP_COLLAB_TMS), copy.BTN_SKIP_COLLAB)
self.assertNotIn("请手动拉相关人员", joined)
self.assertNotIn("若需要咨询内部航线人员", joined)
cards = [x for x in self.extras if x.get("msgtype") == "template_card"]
self.assertTrue(cards)
buttons = [b["text"] for b in cards[-1]["template_card"]["button_list"]]
self.assertEqual(buttons, [copy.BTN_PULL_COLLAB, copy.BTN_SKIP_COLLAB_TMS])
def test_sea_miss_one_card_with_pull_button(self) -> None:
facts = dict(COMPLETE_SEA)
facts["目的港"] = "无价港"
phase = drive_sea(self.flow,
sender_id="s1",
text="海运无价",
reply=self._reply,
injected_facts=facts,
injected_mode="SEA",
)
self.assertEqual(phase, "tms_miss")
sess = self.flow.session_of("s1")
self.assertEqual(sess.status, "询价中")
joined = "\n".join(self.replies)
self.assertIn("暂无匹配报价", joined)
self.assertIn("询价中", joined)
self.assertIn("未查到合适报价", joined)
self.assertIn("是否需要拉内部产品人员进群", joined)
self.assertIn(copy.BTN_PULL_COLLAB, joined)
self.assertNotIn("请手动拉相关人员和智能体到原有群", joined)
cards = [x for x in self.extras if x.get("msgtype") == "template_card"]
self.assertTrue(cards)
buttons = [b["text"] for b in cards[-1]["template_card"].get("button_list") or []]
self.assertEqual(buttons, [copy.BTN_PULL_COLLAB])
def test_sea_tech_fail_is_not_miss(self) -> None:
facts = dict(COMPLETE_SEA)
facts["tms_force"] = "tech"
phase = drive_sea(self.flow,
sender_id="s1",
text="海运技术失败",
reply=self._reply,
injected_facts=facts,
injected_mode="SEA",
)
self.assertEqual(phase, "system_exception")
self.assertIn("TMS接口异常，IT运维排查中", "\n".join(self.replies))
self.assertNotIn(copy.BTN_PULL_COLLAB, "\n".join(self.replies))
tickets = list(self.flow._ledger._tickets.values())
self.assertTrue(tickets)
self.assertEqual(tickets[-1].system_exception, "TMS异常")
class SeaSkipAndPullTests(unittest.TestCase):
def setUp(self) -> None:
os.environ["WECOM_ARCHIVE_SEAT_USER_IDS"] = ""
from agent.config import get_settings
get_settings.cache_clear()
self.ledger = MemoryLedger()
self.ledger.set_quote_templates(
[
{
"templateId": "mem-default-sea",
"name": "海运默认",
"bizType": "海运",
"keywords": [],
"isDefault": True,
}
]
)
self.groups = FakeGroupClient()
self.flow = SeaTextInquiryFlow(self.ledger, group_client=self.groups)
self.replies: list[str] = []
self.extras: list[dict] = []
def _reply(self, text: str, extra=None) -> str:
self.replies.append(text)
self.extras.append(dict(extra) if extra else {})
return ""
def _quote_first(self, sender: str = "s1") -> None:
drive_sea(self.flow,
sender_id=sender,
text="海运有价",
reply=self._reply,
injected_facts=COMPLETE_SEA,
injected_mode="SEA",
)
def test_skip_collab_reuses_air_scheme(self) -> None:
self._quote_first()
phase = self.flow.on_button(
sender_id="s1",
action=copy.BTN_SKIP_COLLAB,
reply=self._reply,
card_meta={"response_code": "rc-sea-1"},
)
self.assertEqual(phase, "wait_adopt")
ticket = self.ledger.get_ticket(work_order_no=self.flow.session_of("s1").work_order_no)
self.assertEqual(ticket.event_wait_kind, "adopt")
self.assertTrue(ticket.event_wait_started_at)
self.assertIn("已跳过内部协同", "\n".join(self.replies))
scheme = "\n".join(x for x in self.replies if "报价方案" in x)
self.assertIn("运输方式：海运", scheme)
self.assertIn("货物数量：", scheme)
self.assertIn("整柜或拼柜：", scheme)
self.assertIn("箱型箱量：", scheme)
self.assertNotIn("件数：", scheme)
self.assertNotIn("重量：", scheme)
self.assertNotIn("体积：", scheme)
self.assertNotIn("包装类型：", scheme)
cards = [
x.get("template_card") or {}
for x in self.extras
if (x.get("template_card") or {}).get("main_title", {}).get("title", "").endswith("报价方案")
]
self.assertTrue(cards)
rows = {r["keyname"]: r["value"] for r in cards[0].get("horizontal_content_list") or []}
self.assertIn("海运", rows.get("航线", ""))
self.assertIn("箱型", rows)
self.assertNotIn("包装", rows)
again = self.flow.on_button(
sender_id="s1", action=copy.BTN_SKIP_COLLAB, reply=self._reply
)
self.assertEqual(again, "card_discard")
def test_pull_group_success_greys_buttons(self) -> None:
self.ledger.set_sea_staff(
[
{
"name": "张产品",
"wecomId": "prod1",
"roleCode": "sea",
"routes": "海运·全部线路",
"status": "active",
}
]
)
self._quote_first()
wo = self.flow.session_of("s1").work_order_no
phase = self.flow.on_button(
sender_id="s1",
action=copy.BTN_PULL_COLLAB,
reply=self._reply,
card_meta={"response_code": "rc-pull-1"},
)
self.assertEqual(phase, "collab_group")
ticket = self.ledger.get_ticket(work_order_no=wo)
self.assertEqual(ticket.event_wait_kind, "product_quote")
self.assertTrue(ticket.event_wait_started_at)
joined = "\n".join(self.replies)
self.assertIn("协同群已创建", joined)
self.assertIn(wo, joined)
self.assertIn("销售", joined)
self.assertIn("产品张产品", joined)
self.assertEqual(self.groups.created[0]["name"], wo)
self.assertIn("prod1", self.groups.created[0]["userids"])
self.assertIn("s1", self.groups.created[0]["userids"])
self.assertNotIn("AI001", self.groups.created[0]["userids"])
self.assertTrue(self.groups.sent)
group_text = self.groups.sent[0]["content"]
self.assertTrue(any(x.get("kind") == "file" for x in self.groups.sent))
self.assertIn(wo, group_text)
self.assertIn("起运港", group_text)
self.assertIn("贸易条款", group_text)
collab_block = group_text.split("协同补充字段：", 1)[-1].split("TMS报价：", 1)[0]
self.assertNotIn("贸易条款", collab_block)
self.assertIn("货好时间", collab_block)
self.assertIn("商品海关编码", group_text)
self.assertNotIn("HS编码", group_text)
self.assertIn("TMS", group_text)
sess = self.flow.session_of("s1")
self.assertNotIn(copy.BTN_PULL_COLLAB, sess.allowed)
self.assertNotIn(copy.BTN_SKIP_COLLAB, sess.allowed)
again = self.flow.on_button(
sender_id="s1", action=copy.BTN_SKIP_COLLAB, reply=self._reply
)
self.assertEqual(again, "card_discard")
def test_pull_prefills_spoken_no_battery(self) -> None:
self.ledger.set_sea_staff(
[
{
"name": "张产品",
"wecomId": "prod1",
"roleCode": "sea",
"routes": "海运·全部线路",
"status": "active",
}
]
)
facts = dict(COMPLETE_SEA)
facts["是否含电"] = "否"
drive_sea(self.flow,
sender_id="s1",
text="海运不含电",
reply=self._reply,
injected_facts=facts,
injected_mode="SEA",
)
private = "\n".join(self.replies)
self.assertNotIn("是否含电", private)
self.flow.on_button(
sender_id="s1",
action=copy.BTN_PULL_COLLAB,
reply=self._reply,
)
group_text = "\n".join(
x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text"
)
self.assertIn("是否含电：否", group_text)
def test_pull_sends_sheet_when_client_not_injected(self) -> None:
"""测服正式路径不注入 group_client，拉群仍要发出填好的模板。"""
from unittest.mock import patch
live = FakeGroupClient()
self.ledger.set_sea_staff(
[
{
"name": "张产品",
"wecomId": "prod1",
"roleCode": "sea",
"routes": "海运·全部线路",
"status": "active",
}
]
)
with patch("agent.policy.sea_text_flow._default_group_client", return_value=live):
flow = SeaTextInquiryFlow(self.ledger, group_client=None)
drive_sea(flow,
sender_id="s9",
text="海运有价",
reply=self._reply,
injected_facts=COMPLETE_SEA,
injected_mode="SEA",
)
phase = flow.on_button(
sender_id="s9",
action=copy.BTN_PULL_COLLAB,
reply=self._reply,
)
self.assertEqual(phase, "collab_group")
self.assertTrue(any(x.get("kind") == "text" for x in live.sent))
self.assertTrue(any(x.get("kind") == "file" for x in live.sent))
def test_pull_group_includes_archive_bot(self) -> None:
self.ledger.upsert_staff(
{
"name": "询价机器人",
"wecomId": "AI001",
"roleCode": "sales",
"status": "active",
}
)
self.ledger.set_sea_staff(
[
{
"name": "张产品",
"wecomId": "prod1",
"roleCode": "sea",
"routes": "海运全线",
"status": "active",
}
]
)
self._quote_first()
self.flow.on_button(
sender_id="s1",
action=copy.BTN_PULL_COLLAB,
reply=self._reply,
)
userids = self.groups.created[0]["userids"]
self.assertIn("AI001", userids)
self.assertIn("s1", userids)
self.assertIn("prod1", userids)
joined = "\n".join(self.replies)
self.assertIn("产品张产品", joined)
self.assertNotIn("产品询价机器人", joined)
self.assertNotIn("销售询价机器人", joined)
def test_pull_no_staff_keeps_button(self) -> None:
self.ledger.set_sea_staff([])
self._quote_first()
phase = self.flow.on_button(
sender_id="s1",
action=copy.BTN_PULL_COLLAB,
reply=self._reply,
)
self.assertEqual(phase, "wait_collab")
self.assertIn("没有可拉的产品人员", "\n".join(self.replies))
self.assertEqual(self.groups.created, [])
self.assertIn(copy.BTN_PULL_COLLAB, self.flow.session_of("s1").allowed)
def test_pull_already_has_group_no_second(self) -> None:
self.ledger.set_sea_staff(
[
{
"name": "李产品",
"wecomId": "prod2",
"roleCode": "sea",
"routes": "上海",
"status": "active",
}
]
)
self._quote_first()
self.flow.on_button(
sender_id="s1",
action=copy.BTN_PULL_COLLAB,
reply=self._reply,
)
n = len(self.groups.created)
self.flow.session_of("s1").allowed = (copy.BTN_PULL_COLLAB,)
phase = self.flow.on_button(
sender_id="s1",
action=copy.BTN_PULL_COLLAB,
reply=self._reply,
)
self.assertEqual(phase, "collab_group")
self.assertEqual(len(self.groups.created), n)
self.assertIn("群已在", "\n".join(self.replies))
def test_miss_group_message_writes_no_quote(self) -> None:
self.ledger.set_sea_staff(
[
{
"name": "王产品",
"wecomId": "prod3",
"roleCode": "sea",
"routes": "海运·全部线路",
"status": "active",
}
]
)
facts = dict(COMPLETE_SEA)
facts["目的港"] = "无价港"
drive_sea(self.flow,
sender_id="s1",
text="海运无价",
reply=self._reply,
injected_facts=facts,
injected_mode="SEA",
)
self.flow.on_button(
sender_id="s1",
action=copy.BTN_PULL_COLLAB,
reply=self._reply,
)
self.assertTrue(self.groups.sent)
group_text = self.groups.sent[0]["content"]
self.assertIn("TMS 暂无匹配报价", group_text)
self.assertNotIn("未查到合适报价", group_text)
self.assertIn("\n\n协同补充字段：\n", group_text)
self.assertIn("\n\nTMS报价：\n", group_text)
self.assertNotIn("确认报价", group_text)
class SeaGroupCollabFieldTests(unittest.TestCase):
def setUp(self) -> None:
self.ledger = MemoryLedger()
self.ledger.set_quote_templates(
[
{
"templateId": "mem-default-sea",
"name": "海运默认",
"bizType": "海运",
"keywords": [],
"isDefault": True,
}
]
)
self.groups = FakeGroupClient()
self.flow = SeaTextInquiryFlow(self.ledger, group_client=self.groups)
self.replies: list[str] = []
def _reply(self, text: str, extra=None) -> None:
self.replies.append(text)
def test_group_text_ignores_trade_terms_saves_ready_date(self) -> None:
self.ledger.set_sea_staff(
[
{
"name": "赵产品",
"wecomId": "prod4",
"roleCode": "sea",
"routes": "海运·全部线路",
"status": "active",
}
]
)
drive_sea(self.flow,
sender_id="sales1",
text="海运有价",
reply=self._reply,
injected_facts=COMPLETE_SEA,
injected_mode="SEA",
)
self.flow.on_button(
sender_id="sales1",
action=copy.BTN_PULL_COLLAB,
reply=self._reply,
)
wo = self.flow.session_of("sales1").work_order_no
ticket = self.ledger.get_ticket(work_order_no=wo)
chat_id = ticket.collab_chat_id
n = len(self.replies)
ignored = handle_group_collab(
InboundMessage(
sender_id="prod4",
message_id="g1",
content="贸易条款：CIF",
chat_id=chat_id,
chat_type="group",
),
flow=self.flow,
ledger=self.ledger,
)
self.assertEqual(ignored, "collab_not_recognized")
self.assertFalse(ticket.collab_facts.get("贸易条款"))
phase = handle_group_collab(
InboundMessage(
sender_id="prod4",
message_id="g2",
content="货好时间：下周",
chat_id=chat_id,
chat_type="group",
),
flow=self.flow,
ledger=self.ledger,
)
self.assertEqual(phase, "collab_field_saved")
self.assertEqual(len(self.replies), n)
self.assertEqual(ticket.collab_facts.get("货好时间"), "下周")
def test_group_price_chat_is_ignored(self) -> None:
self.ledger.set_sea_staff(
[
{
"name": "赵产品",
"wecomId": "prod4",
"roleCode": "sea",
"routes": "海运·全部线路",
"status": "active",
}
]
)
drive_sea(self.flow,
sender_id="sales1",
text="海运有价",
reply=self._reply,
injected_facts=COMPLETE_SEA,
injected_mode="SEA",
)
self.flow.on_button(
sender_id="sales1",
action=copy.BTN_PULL_COLLAB,
reply=self._reply,
)
wo = self.flow.session_of("sales1").work_order_no
ticket = self.ledger.get_ticket(work_order_no=wo)
phase = handle_group_collab(
InboundMessage(
sender_id="prod4",
message_id="g2",
content="报价 1200 美元可以做",
chat_id=ticket.collab_chat_id,
chat_type="group",
),
flow=self.flow,
ledger=self.ledger,
)
self.assertEqual(phase, "wait_tms_choice")
self.assertNotEqual(str(ticket.quote.get("total") or ""), "1200")
texts = [x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text"]
self.assertTrue(any("在TMS基础上改" in t and "不用TMS报价" in t for t in texts))
phase = handle_group_collab(
InboundMessage(
sender_id="prod4",
message_id="g2b",
content="按TMS改一下就行",
chat_id=ticket.collab_chat_id,
chat_type="group",
),
flow=self.flow,
ledger=self.ledger,
)
self.assertEqual(phase, "product_quote_sent")
ticket = self.ledger.get_ticket(work_order_no=wo)
names = [str(x.get("name") or "") for x in (ticket.quote.get("fee_rows") or ticket.quote.get("fee_lines") or [])]
self.assertTrue(any("空运费" in n or "海运费" in n or "附加" in n for n in names) or "1280" in str(ticket.quote.get("total") or ""))
deal_texts = [x["content"] for x in self.groups.sent if x.get("kind") == "text" and "成交" in (x.get("content") or "")]
self.assertTrue(deal_texts)
def test_keep_tms_reads_json_string_fee_lines(self) -> None:
"""TMS fee_lines 常是 JSON 字符串，在TMS基础上改不能当成没明细。"""
from agent.policy.sea_group_ops import merge_quote_fees
base = {
"source_label": "TMS 标准报价",
"total": "2287.65",
"currency": "CNY",
"fee_lines": (
'[{"name":"海运费","amount":"100","currency":"USD","unit_price":"100"},'
'{"name":"操作费","amount":"680","currency":"CNY","unit_price":"680"},'
'{"name":"文件费","amount":"450","currency":"CNY","unit_price":"450"}]'
),
}
overlay = {
"source_label": "产品报价",
"total": "1200",
"currency": "CNY",
"fee_rows": [
{"name": "操作费", "amount": "700", "currency": "CNY"},
{"name": "文件发", "amount": "500", "currency": "CNY"},
],
}
merged = merge_quote_fees(base, overlay)
by_name = {str(x.get("name") or ""): x for x in (merged.get("fee_rows") or [])}
self.assertEqual(str(by_name["海运费"].get("amount") or ""), "100")
self.assertEqual(str(by_name["操作费"].get("amount") or ""), "700")
self.assertEqual(str(by_name["操作费"].get("unit_price") or ""), "700")
self.assertEqual(str(by_name["文件费"].get("amount") or ""), "500")
self.assertEqual(str(by_name["文件费"].get("unit_price") or ""), "500")
self.assertEqual(str(by_name["海运费"].get("unit_price") or ""), "100")
self.assertNotIn("文件发", by_name)
self.assertEqual(str(merged.get("total") or ""), "2357.65")
self.assertNotEqual(str(merged.get("total") or ""), "1200")
def test_usd_prefix_fee_lines_are_both_kept(self) -> None:
"""海运费：USD 100.00 / 燃油附加费：USD 15.00 两行都要抽出，不能只剩海运费。"""
from agent.llm.extract_text import extract_product_quote
quote = extract_product_quote("海运费：USD 100.00\n燃油附加费：USD 15.00")
rows = {str(x.get("name") or ""): x for x in (quote.get("fee_rows") or [])}
self.assertIn("海运费", rows)
self.assertIn("燃油附加费", rows)
self.assertEqual(str(rows["海运费"].get("amount") or ""), "100.00")
self.assertEqual(str(rows["燃油附加费"].get("amount") or ""), "15.00")
def test_land_weight_band_unit_price_and_rates(self) -> None:
"""陆运集拼：2-500KG 单价 + 燃油附加费率/税率，不能把档下限 2 当成燃油附加费。"""
from agent.llm.extract_text import extract_product_quote
quote = extract_product_quote(
"2-500KG 单价CNY990，燃油附加费率1%，税率6%"
)
self.assertEqual(str(quote.get("total") or ""), "990")
self.assertEqual(str(quote.get("fuel_surcharge_rate") or ""), "1%")
self.assertEqual(str(quote.get("tax_rate") or ""), "6%")
tiers = list(quote.get("weightPrices") or [])
self.assertEqual(len(tiers), 1)
self.assertEqual(int(tiers[0].get("min") or 0), 2)
self.assertEqual(int(tiers[0].get("max") or 0), 500)
self.assertEqual(str(tiers[0].get("price") or ""), "990")
rows = {str(x.get("name") or ""): x for x in (quote.get("fee_rows") or [])}
self.assertIn("2-500KG", rows)
self.assertEqual(str(rows["2-500KG"].get("amount") or ""), "990")
self.assertNotEqual(str(rows.get("燃油附加费", {}).get("amount") or ""), "2")
self.assertEqual(str(rows.get("燃油附加费率", {}).get("amount") or ""), "1%")
self.assertEqual(str(rows.get("税率", {}).get("amount") or ""), "6%")
def test_air_unit_qty_amount_keeps_air_freight_name(self) -> None:
"""空运费 单价/数量/金额：费用名是空运费，金额用「金额」不是单价。"""
from agent.llm.extract_text import extract_product_quote
quote = extract_product_quote(
"空运段，空运费 单价CNY99.00，数量100 ，金额CNY199.00@询价小助手"
)
rows = {str(x.get("name") or ""): x for x in (quote.get("fee_rows") or [])}
self.assertIn("空运费", rows)
self.assertNotIn("海运费", rows)
self.assertEqual(str(rows["空运费"].get("amount") or ""), "199.00")
self.assertEqual(str(rows["空运费"].get("currency") or ""), "CNY")
self.assertEqual(str(rows["空运费"].get("unit_price") or ""), "99.00")
self.assertEqual(str(quote.get("total") or ""), "199.0")
def test_fee_line_text_is_a_product_quote(self) -> None:
"""产品按行改价：码头操作费100CNY / 报关费改为300CNY，必须抽出费用名，不能只当海运费合计。"""
from agent.llm.extract_text import extract_product_quote
quote = extract_product_quote("码头操作费100CNY\n报关费改为300CNY")
rows = {str(x.get("name") or ""): x for x in (quote.get("fee_rows") or [])}
self.assertIn("码头操作费", rows)
self.assertIn("报关费", rows)
self.assertEqual(str(rows["码头操作费"].get("amount") or ""), "100")
self.assertEqual(str(rows["报关费"].get("amount") or ""), "300")
self.assertNotEqual(
[str(x.get("name") or "") for x in (quote.get("fee_rows") or [])],
["海运费"],
)
def test_fee_change_typo_and_colon(self) -> None:
"""操作费改为700 + 文件发改为：500，两行都要抽出，文件发要对上文件费。"""
from agent.llm.extract_text import extract_product_quote
from agent.policy.sea_group_ops import _fee_name_same, merge_quote_fees
quote = extract_product_quote("操作费改为700CNY\n文件发改为：500CNY")
rows = {str(x.get("name") or ""): x for x in (quote.get("fee_rows") or [])}
self.assertEqual(str(rows["操作费"].get("amount") or ""), "700")
self.assertEqual(str(rows["文件发"].get("amount") or ""), "500")
self.assertTrue(_fee_name_same("文件发", "文件费"))
self.assertFalse(_fee_name_same("操作费", "码头操作费"))
merged = merge_quote_fees(
{
"source_label": "TMS标准报价",
"fee_rows": [
{"name": "操作费", "amount": "680", "currency": "CNY"},
{"name": "文件费", "amount": "450", "currency": "CNY"},
],
},
quote,
)
by_name = {str(x.get("name") or ""): x for x in (merged.get("fee_rows") or [])}
self.assertEqual(str(by_name["操作费"].get("amount") or ""), "700")
self.assertEqual(str(by_name["文件费"].get("amount") or ""), "500")
self.assertNotIn("文件发", by_name)
class SeaContinueTests(unittest.TestCase):
def setUp(self) -> None:
from agent.redis_coord.runtime import shutdown_redis_runtime
shutdown_redis_runtime()
self.ledger = MemoryLedger()
self.ledger.set_quote_templates(
[
{
"templateId": "mem-default-sea",
"name": "海运默认",
"bizType": "海运",
"keywords": [],
"isDefault": True,
}
]
)
self.flow = SeaTextInquiryFlow(self.ledger)
self.replies: list[str] = []
self.extras: list[dict] = []
def _reply(self, text: str, extra=None) -> str:
self.replies.append(text)
self.extras.append(dict(extra) if extra else {})
return ""
def test_continue_after_skip_replays_sea_scheme_not_air(self) -> None:
"""已跳过协同后再续办：必须海运方案卡，不能套空运件数包装。"""
from unittest.mock import patch
from agent.channel.queue import MemoryMessageStore
from agent.handlers.continue_thread import handle_continue_thread
from agent.policy.air_text_flow import AirTextInquiryFlow
drive_sea(self.flow,
sender_id="c1",
text="海运有价",
reply=self._reply,
injected_facts=COMPLETE_SEA,
injected_mode="SEA",
)
self.flow.on_button(
sender_id="c1",
action=copy.BTN_SKIP_COLLAB,
reply=self._reply,
card_meta={"response_code": "rc-cont-1"},
)
wo = self.flow.session_of("c1").work_order_no
store = MemoryMessageStore()
msg = InboundMessage(
sender_id="c1",
message_id="m-cont-sea",
content=f"继续工单{wo}",
)
air_trap = AirTextInquiryFlow(MemoryLedger())
with patch(
"agent.handlers.continue_thread.get_sea_text_flow", return_value=self.flow
), patch(
"agent.handlers.continue_thread.get_air_text_flow", return_value=air_trap
):
phase = handle_continue_thread(msg, thread_id=wo, store=store)
self.assertEqual(phase, "wait_adopt")
cards = [
(item.payload or {}).get("template_card") or {}
for item in store._outbox.values()
]
titles = [
(c.get("main_title") or {}).get("title") or "" for c in cards if c
]
self.assertTrue(any(t == f"{wo}报价方案" for t in titles))
rows = {}
for card in cards:
if (card.get("main_title") or {}).get("title") == f"{wo}报价方案":
rows = {
r["keyname"]: r["value"]
for r in card.get("horizontal_content_list") or []
}
self.assertIn("海运", rows.get("航线", ""))
self.assertIn("箱型", rows)
self.assertNotIn("包装", rows)
def test_continue_from_ledger_uses_sea_collab_card(self) -> None:
"""进程无书签：主账海运已报价应回收海运协同卡，不是空运卡。"""
from unittest.mock import patch
from agent.channel.queue import MemoryMessageStore
from agent.handlers.continue_thread import handle_continue_thread
from agent.policy.air_text_flow import AirTextInquiryFlow
created = self.ledger.create_ticket(
sender_id="c2", business_line="SEA", facts=COMPLETE_SEA
)
wo = created["work_order_no"]
self.ledger.upsert_quote(
work_order_no=wo,
quote={
"total": "USD 1,200",
"eta": "18天",
"valid_until": "2026-09-20",
"source_label": "TMS 标准报价",
},
to_status="已报价",
)
empty = SeaTextInquiryFlow(self.ledger)
store = MemoryMessageStore()
msg = InboundMessage(
sender_id="c2",
message_id="m-cont-ledger",
content=f"继续工单{wo}",
)
air_trap = AirTextInquiryFlow(MemoryLedger())
with patch(
"agent.handlers.continue_thread.get_sea_text_flow", return_value=empty
), patch(
"agent.handlers.continue_thread.get_air_text_flow", return_value=air_trap
):
phase = handle_continue_thread(msg, thread_id=wo, store=store)
self.assertEqual(phase, "wait_collab")
sess = empty.session_by_ticket("c2", wo)
self.assertIsNotNone(sess)
self.assertEqual(sess.business_line, "SEA")
joined = "\n".join(item.content for item in store._outbox.values())
self.assertIn(copy.BTN_PULL_COLLAB, joined)
self.assertIn("是否拉产品人员进群协同", joined)
self.assertNotIn("包装方式", joined)
def test_continue_after_quote_file_returns_deal(self) -> None:
"""报价单已发出后再续办：必须成交跟进，不能退回生成报价单。"""
from unittest.mock import patch
from agent.channel.queue import MemoryMessageStore
from agent.handlers.continue_thread import handle_continue_thread
from agent.policy.air_text_flow import AirTextInquiryFlow
drive_sea(self.flow,
sender_id="c3",
text="海运有价",
reply=self._reply,
injected_facts=COMPLETE_SEA,
injected_mode="SEA",
)
self.flow.on_button(
sender_id="c3",
action=copy.BTN_SKIP_COLLAB,
reply=self._reply,
card_meta={"response_code": "rc-deal-1"},
)
wo = self.flow.session_of("c3").work_order_no
self.flow.on_button(
sender_id="c3",
action=copy.BTN_ADOPT_EXCEL,
reply=self._reply,
card_meta={"event_key": f"adopt_excel:{wo}", "response_code": "rc-xl"},
)
self.assertEqual(self.flow.session_of("c3").phase, "wait_deal")
file_extras = [x for x in self.extras if x.get("msgtype") == "file"]
self.assertTrue(file_extras)
follow = file_extras[-1].get("followup_card") or {}
self.assertTrue(str(follow.get("task_id") or "").startswith("deal-"))
separate_deal = [
x
for x in self.extras
if x.get("msgtype") == "template_card"
and str((x.get("template_card") or {}).get("task_id") or "").startswith("deal-")
]
self.assertEqual(separate_deal, [])
empty = SeaTextInquiryFlow(self.ledger)
store = MemoryMessageStore()
msg = InboundMessage(
sender_id="c3",
message_id="m-cont-deal",
content=f"继续工单{wo}",
)
air_trap = AirTextInquiryFlow(MemoryLedger())
with patch(
"agent.handlers.continue_thread.get_sea_text_flow", return_value=empty
), patch(
"agent.handlers.continue_thread.get_air_text_flow", return_value=air_trap
):
phase = handle_continue_thread(msg, thread_id=wo, store=store)
self.assertEqual(phase, "wait_deal")
joined = "\n".join(item.content for item in store._outbox.values())
self.assertFalse(any(x in joined for x in (copy.BTN_DEAL, "成交")))
self.assertNotIn(copy.BTN_SKIP_COLLAB, joined)
class SeaGroupCopyTests(unittest.TestCase):
def test_group_brief_uses_customs_code_and_section_breaks(self) -> None:
text = copy.sea_group_brief(
work_order_no="WO202609150023",
facts={
"起运港": "纽约港",
"目的港": "巴生西港",
"品名": "机器设备",
"货量": "15",
"整柜或拼柜": "整柜",
"箱型箱量": "40HQ*1",
"报价日期": "2026-09-15",
},
quote=None,
collab_facts={},
sales_name="吴季林",
product_names=["罗小华"],
has_price=False,
)
self.assertIn("@吴季林 @罗小华", text)
self.assertIn("询价字段：\n运输方式：海运", text)
self.assertIn("\n\n协同补充字段：\n", text)
self.assertIn("商品海关编码：-", text)
self.assertIn("货好时间：-", text)
collab_block = text.split("协同补充字段：", 1)[-1].split("TMS报价：", 1)[0]
self.assertNotIn("贸易条款", collab_block)
self.assertNotIn("HS编码", text)
self.assertIn("\n\nTMS报价：\nTMS 暂无匹配报价", text)
self.assertIn("产品人员可通过“纯文字”告知新报价，或直接将新报价excel表发给智能体。", text)
self.assertIn("附件报价单仅作为文件存储于后台，智能体不读取且不将报价内容写进系统", text)
self.assertNotIn("航线人员可通过", text)
self.assertNotIn("未查到合适报价", text)
self.assertNotIn("确认报价", text)
def test_group_brief_has_confirm_hint_when_tms_priced(self) -> None:
text = copy.sea_group_brief(
work_order_no="WO202609160004",
facts={"起运港": "DONGGUAN", "目的港": "海防", "品名": "机器设备"},
quote={"total": "2287.65", "currency": "CNY", "valid_until": "2026-09-30"},
collab_facts={},
sales_name="吴季林",
product_names=["林绵勋"],
has_price=True,
)
self.assertIn("报价有效期：2026-09-30", text)
self.assertIn("请产品人员回复“确认报价”确定采用TMS报价。", text)
self.assertIn("亦可通过“纯文字”告知新报价，或直接将新报价excel表发给智能体。", text)
self.assertIn("附件报价单仅作为文件存储于后台，智能体不读取且不将报价内容写进系统", text)
self.assertNotIn("请航线人员回复", text)
self.assertNotIn("人工报价：", text)
def test_group_brief_manual_quote_stays_off_tms(self) -> None:
text = copy.sea_group_brief(
work_order_no="WO202609160006",
facts={"起运港": "上海港", "目的港": "海防", "品名": "机器设备"},
quote={
"source_label": "产品报价",
"total": "115",
"currency": "USD",
"fee_rows": [
{"name": "海运费", "amount": "100.00", "currency": "USD"},
{"name": "燃油附加费", "amount": "15.00", "currency": "USD"},
],
},
collab_facts={},
sales_name="吴季林",
product_names=["林绵勋", "罗小华", "吴季林"],
has_price=True,
tms_quote={},
product_quote={
"source_label": "产品报价",
"fee_rows": [
{"name": "海运费", "amount": "100.00", "currency": "USD"},
{"name": "燃油附加费", "amount": "15.00", "currency": "USD"},
],
},
)
self.assertIn("TMS报价：\nTMS 暂无匹配报价", text)
self.assertIn("产品人员可通过“纯文字”告知新报价，或直接将新报价excel表发给智能体。", text)
self.assertIn("人工报价：\n海运费：USD 100.00\n燃油附加费：USD 15.00", text)
self.assertNotIn("确认报价", text)
self.assertNotIn("总销售价合计：USD 100.00", text)
def test_scheme_card_uses_sea_fields(self) -> None:
facts = {
"起运港": "上海",
"目的港": "洛杉矶",
"品名": "普货",
"货量": "15件",
"整柜或拼柜": "整柜",
"箱型箱量": "1x40HQ",
"报价日期": "2026-09-15",
}
quote = {"total": "USD 1200", "eta": "18天", "valid_until": "2026-09-20"}
text = copy.scheme_card(
work_order_no="WO202609150021",
status="已报价",
facts=facts,
quote=quote,
transport_mode="海运",
)
self.assertIn("运输方式：海运", text)
self.assertIn("货物数量：15件", text)
self.assertIn("箱型箱量：1x40HQ", text)
self.assertNotIn("件数：", text)
self.assertNotIn("包装类型：", text)
rows = {
x["keyname"]: x["value"]
for x in copy.scheme_horizontal_rows(
work_order_no="WO202609150021",
facts=facts,
quote=quote,
transport_mode="海运",
)
}
self.assertIn("海运 上海→洛杉矶", rows["航线"])
self.assertIn("15件", rows["货物"])
self.assertIn("1x40HQ", rows["箱型"])
self.assertNotIn("包装", rows)
def test_lost_ack_asks_reason_only(self) -> None:
text = copy.deal_ack("未成交", "WO202609160004")
self.assertEqual(text, "已记录为「未成交」。请补充未成交原因。")
self.assertNotIn("将关闭智能体跟进", text)
def test_invited_names_carry_role(self) -> None:
text = copy.sea_group_created(
work_order_no="WO202609150019",
sales_name="吴季林",
product_names=["罗小华"],
)
self.assertEqual(
text,
"协同群已创建，群名与工单号一致：WO202609150019。"
"已邀请：销售吴季林、产品罗小华。请在群聊中继续确认报价。",
)
def test_group_only_excel_pdf_copy(self) -> None:
self.assertEqual(
copy.group_only_excel_pdf(),
"群里只收 Excel 和 PDF，其它文件不会存入工单",
)
class SeaGroupOpsTests(unittest.TestCase):
"""群协同：附件类型、产品报价、解散后再拉、终态提醒。"""
def setUp(self) -> None:
self.ledger = MemoryLedger()
self.ledger.set_quote_templates(
[
{
"templateId": "mem-default-sea",
"name": "海运默认",
"bizType": "海运",
"keywords": [],
"isDefault": True,
}
]
)
self.ledger.set_sea_staff(
[
{
"name": "赵产品",
"wecomId": "prod4",
"roleCode": "sea",
"routes": "海运·全部线路",
"status": "active",
}
]
)
self.ledger.upsert_staff({"name": "吴吉林", "wecomId": "sales1"})
self.groups = FakeGroupClient()
self.flow = SeaTextInquiryFlow(self.ledger, group_client=self.groups)
self.replies: list[str] = []
def _reply(self, text: str, extra=None) -> None:
self.replies.append(text)
def _open_group(self) -> tuple[str, str]:
drive_sea(self.flow,
sender_id="sales1",
text="海运有价",
reply=self._reply,
injected_facts=COMPLETE_SEA,
injected_mode="SEA",
)
self.flow.on_button(
sender_id="sales1",
action=copy.BTN_PULL_COLLAB,
reply=self._reply,
)
sess = self.flow.session_of("sales1")
return sess.work_order_no, sess.collab_chat_id
def test_group_rejects_non_excel_pdf(self) -> None:
wo, chat = self._open_group()
phase = handle_group_collab(
InboundMessage(
sender_id="sales1",
message_id="img1",
content="",
msg_type="image",
chat_id=chat,
chat_type="group",
media={"filename": "现场.jpg", "media_id": "mid-1"},
),
flow=self.flow,
ledger=self.ledger,
)
self.assertEqual(phase, "file_type_rejected")
texts = [x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text"]
self.assertIn("群里只收 Excel 和 PDF，其它文件不会存入工单", texts)
ticket = self.ledger.get_ticket(work_order_no=wo)
self.assertFalse(ticket.attachments)
def test_sales_excel_is_archived_not_quote(self) -> None:
wo, chat = self._open_group()
phase = handle_group_collab(
InboundMessage(
sender_id="sales1",
message_id="xls1",
content="",
msg_type="file",
chat_id=chat,
chat_type="group",
media={"filename": "对账单.xlsx", "media_id": "mid-2"},
),
flow=self.flow,
ledger=self.ledger,
)
self.assertEqual(phase, "file_archived")
ticket = self.ledger.get_ticket(work_order_no=wo)
self.assertEqual(ticket.attachments[0]["file_name"], "对账单.xlsx")
self.assertNotEqual(ticket.quote.get("source_label"), "产品附件报价")
def test_product_excel_quotes_and_deal(self) -> None:
wo, chat = self._open_group()
phase = handle_group_collab(
InboundMessage(
sender_id="prod4",
message_id="xls2",
content="",
msg_type="file",
chat_id=chat,
chat_type="group",
media={"filename": "产品报价.pdf", "media_id": "mid-3"},
),
flow=self.flow,
ledger=self.ledger,
)
self.assertEqual(phase, "product_file_quote")
ticket = self.ledger.get_ticket(work_order_no=wo)
self.assertEqual(ticket.event_wait_kind, "deal")
self.assertTrue(ticket.event_wait_started_at)
self.assertEqual(ticket.attachments[0]["file_name"], "产品报价.pdf")
self.assertEqual(ticket.status, "已报价")
texts = [x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text"]
self.assertIn(f"工单{wo}的1个附件已保存到后台工单", texts)
self.assertNotIn("附件已保存后台", texts)
ask = [t for t in texts if "在TMS基础上改" in t]
self.assertFalse(ask)
self.assertTrue(any(x.get("kind") == "card" for x in self.groups.sent))
def test_terminal_file_speaks(self) -> None:
wo, chat = self._open_group()
self.ledger.transition(work_order_no=wo, to_status="已成交")
sess = self.flow.session_of("sales1")
sess.status = "已成交"
phase = handle_group_collab(
InboundMessage(
sender_id="prod4",
message_id="xls3",
content="",
msg_type="file",
chat_id=chat,
chat_type="group",
media={"filename": "晚到.xlsx", "media_id": "mid-4"},
),
flow=self.flow,
ledger=self.ledger,
)
self.assertEqual(phase, "ticket_locked_file")
texts = [x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text"]
self.assertTrue(any("不能再上传附件" in t for t in texts))
self.assertFalse(self.ledger.get_ticket(work_order_no=wo).attachments)
def test_handoff_group_stays_silent(self) -> None:
from agent.routing.group_intent import INTENT_FIELDS
wo, chat = self._open_group()
self.ledger.transition(work_order_no=wo, to_status="转人工")
sess = self.flow.session_of("sales1")
sess.status = "转人工"
sess.handoff_committed = True
self.flow._save(sess)
before = len(self.groups.sent)
phase = handle_group_collab(
InboundMessage(
sender_id="prod4",
message_id="h1",
content="客户名称张三",
chat_id=chat,
chat_type="group",
),
flow=self.flow,
ledger=self.ledger,
injected_intent=INTENT_FIELDS,
)
self.assertEqual(phase, "handoff_ignore")
after = [x.get("content") or "" for x in self.groups.sent[before:] if x.get("kind") == "text"]
self.assertFalse(any("不能修改工单信息" in t for t in after))
ticket = self.ledger.get_ticket(work_order_no=wo)
self.assertNotIn("客户名称", ticket.collab_facts or {})
def test_dissolve_then_cannot_recreate(self) -> None:
wo, chat = self._open_group()
phase = handle_group_collab(
InboundMessage(
sender_id="sales1",
message_id="dis1",
content="这个群解散吧",
chat_id=chat,
chat_type="group",
),
flow=self.flow,
ledger=self.ledger,
injected_intent="dissolve",
)
self.assertEqual(phase, "group_dissolved")
self.assertTrue(any(x.get("kind") == "dismiss" for x in self.groups.sent))
self.assertTrue(self.ledger.get_ticket(work_order_no=wo).collab_ended)
sess = self.flow.session_of("sales1")
sess.allowed = (copy.BTN_PULL_COLLAB,)
again = self.flow.on_button(
sender_id="sales1",
action=copy.BTN_PULL_COLLAB,
reply=self._reply,
)
self.assertEqual(again, "collab_ended")
self.assertIn("不能再拉群", "\n".join(self.replies))
def test_group_oral_keeps_inquiry_facts_when_snapshot_skinny(self) -> None:
wo, chat = self._open_group()
real_find = self.ledger.find_by_collab_chat
def skinny(*, chat_id: str):
ticket = real_find(chat_id=chat_id)
if ticket is None:
return None
return {
"found": True,
"work_order_no": ticket.work_order_no,
"sales_wecom_id": ticket.sales_wecom_id,
"collab_chat_id": ticket.collab_chat_id,
"collab_member_ids": list(ticket.collab_member_ids),
"collab_facts": dict(ticket.collab_facts),
}
self.ledger.find_by_collab_chat = skinny # type: ignore[method-assign]
self.flow._by_ticket.clear()
self.flow._by_sender.clear()
phase = handle_group_collab(
InboundMessage(
sender_id="sales1",
message_id="oral-keep-facts-1",
content="不含电、不含油、随时可提",
chat_id=chat,
chat_type="group",
),
flow=self.flow,
ledger=self.ledger,
injected_intent="collab_fields",
)
self.assertEqual(phase, "collab_field_saved")
texts = "\n".join(x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text")
self.assertIn("上海", texts)
self.assertIn("洛杉矶", texts)
self.assertIn("随时可提", texts)
self.assertNotRegex(texts, r"起运港：\s*-")
self.assertIn("@吴吉林 @赵产品", texts)
self.assertNotRegex(texts, r"^@销售(\s|$)")
briefs = [
x
for x in self.groups.sent
if x.get("kind") == "text" and "询价字段" in (x.get("content") or "")
]
self.assertTrue(briefs)
self.assertIn("sales1", briefs[-1].get("mention_userids") or [])
self.assertIn("prod4", briefs[-1].get("mention_userids") or [])
self.assertNotIn("AI001", briefs[-1].get("mention_userids") or [])
def test_group_oral_oil_power_ready_is_saved(self) -> None:
wo, chat = self._open_group()
n = len([x for x in self.groups.sent if x.get("kind") == "text"])
phase = handle_group_collab(
InboundMessage(
sender_id="sales1",
message_id="oral-collab-1",
content="不含电、不含油、随时可提",
chat_id=chat,
chat_type="group",
),
flow=self.flow,
ledger=self.ledger,
injected_intent="collab_fields",
)
self.assertEqual(phase, "collab_field_saved")
ticket = self.ledger.get_ticket(work_order_no=wo)
self.assertEqual(ticket.collab_facts.get("是否含电"), "否")
self.assertEqual(ticket.collab_facts.get("是否含油"), "否")
self.assertEqual(ticket.collab_facts.get("货好时间"), "随时可提")
texts = [x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text"]
self.assertGreater(len(texts), n)
def test_collab_facts_are_projected_for_quote_sheet(self) -> None:
from agent.policy.quote_overlay import overlay_labeled_facts
from agent.policy.sea_group_ops import _facts_for_fill
from openpyxl import Workbook, load_workbook
filled = _facts_for_fill(
{"起运港": "DONGGUAN", "目的港": "海防"},
{"货好时间": "随时可提", "是否含油": "否", "是否含电": "否", "HS编码": "8479"},
)
self.assertEqual(filled["cargo_ready_date"], "随时可提")
self.assertEqual(filled["hs_code"], "8479")
self.assertIn("油：否", filled["oil_elec_mag"])
self.assertIn("电：否", filled["oil_elec_mag"])
import tempfile
from pathlib import Path
wb = Workbook()
ws = wb.active
ws["A1"] = "*货好时间/Cargo Ready Date："
ws["A2"] = "*商品海关编码/HS Code："
ws["A3"] = "*是否含油,电,磁/OIL,ELEC,MAG："
path = str(Path(tempfile.mkdtemp(prefix="collab-sheet-")) / "src.xlsx")
wb.save(path)
out = overlay_labeled_facts(filepath=path, facts=filled)
painted = load_workbook(out["filepath"])
sheet = painted.active
self.assertEqual(sheet["B1"].value, "随时可提")
self.assertEqual(sheet["B2"].value, "8479")
self.assertIn("油：否", str(sheet["B3"].value or ""))
def test_group_deal_text_asks_to_type(self) -> None:
"""群里成交跟进是文字版：@销售、请确认、请输入，不要报价合计和请选择。"""
text = copy.group_deal_followup(sales_name="吴季林", work_order_no="WO202609160001")
self.assertEqual(
text,
(
"@吴季林\n"
"询价工单 WO202609160001\n"
"报价单已发送，请确认成交状态\n"
"咨询成交状态：请输入成交 / 未成交 (未成交需补充原因)/ 协商中"
),
)
wo, chat = self._open_group()
handle_group_collab(
InboundMessage(
sender_id="prod4",
message_id="xls-deal-copy",
content="",
msg_type="file",
chat_id=chat,
chat_type="group",
media={"filename": "产品报价.pdf", "media_id": "mid-deal-copy"},
),
flow=self.flow,
ledger=self.ledger,
)
texts = [x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text"]
follow = [t for t in texts if "请确认成交状态" in t and "请输入成交" in t]
self.assertTrue(follow)
self.assertTrue(any(t.startswith("@") and f"询价工单 {wo}" in t for t in follow))
self.assertFalse(any("请选择成交" in t or "请跟进成交状态" in t or "总销售价合计" in t for t in follow))
def test_group_text_deal_does_not_go_to_private_card(self) -> None:
"""群里打「成交」不能当私聊点卡丢掉，必须走群成交。"""
from unittest.mock import patch
from agent.routing.decision import RouteDecision
from agent.routing.dispatch import dispatch_inbound
wo, chat = self._open_group()
with patch("agent.handlers.group_collab.get_sea_text_flow", return_value=self.flow):
result = dispatch_inbound(
InboundMessage(
sender_id="sales1",
message_id="deal-dispatch-1",
content="成交",
chat_id=chat,
chat_type="group",
),
RouteDecision(intent="ordinary_text_other"),
)
self.assertEqual(result, "group_collab")
ticket = self.ledger.get_ticket(work_order_no=wo)
self.assertEqual(ticket.status, "已成交")
def test_sales_can_close_deal_after_skinny_snapshot(self) -> None:
"""按群找回的工单只有字段、没有报价时，未成交也要能入账。"""
wo, chat = self._open_group()
real_find = self.ledger.find_by_collab_chat
def skinny(*, chat_id: str):
ticket = real_find(chat_id=chat_id)
if ticket is None:
return None
return {
"found": True,
"work_order_no": ticket.work_order_no,
"sales_wecom_id": ticket.sales_wecom_id,
"collab_chat_id": ticket.collab_chat_id,
"collab_member_ids": list(ticket.collab_member_ids),
"facts": dict(ticket.facts),
}
self.ledger.find_by_collab_chat = skinny # type: ignore[method-assign]
self.flow._by_ticket.clear()
self.flow._by_sender.clear()
phase = handle_group_collab(
InboundMessage(
sender_id="sales1",
message_id="lost-skinny-1",
content="未成交",
chat_id=chat,
chat_type="group",
),
flow=self.flow,
ledger=self.ledger,
)
self.assertEqual(phase, "wait_lost_reason")
self.assertEqual(self.ledger.get_ticket(work_order_no=wo).status, "未成交")
def test_sales_next_text_is_lost_reason(self) -> None:
wo, chat = self._open_group()
handle_group_collab(
InboundMessage(
sender_id="sales1",
message_id="lost-1",
content="未成交",
chat_id=chat,
chat_type="group",
),
flow=self.flow,
ledger=self.ledger,
)
phase = handle_group_collab(
InboundMessage(
sender_id="sales1",
message_id="lost-2",
content="客人比价了",
chat_id=chat,
chat_type="group",
),
flow=self.flow,
ledger=self.ledger,
)
self.assertEqual(phase, "done")
self.assertEqual(self.ledger.get_ticket(work_order_no=wo).lost_reason, "客人比价了")
def test_sales_lost_and_reason_same_message(self) -> None:
"""群里一句同时说未成交和原因，不再追问。"""
wo, chat = self._open_group()
phase = handle_group_collab(
InboundMessage(
sender_id="sales1",
message_id="lost-inline-1",
content="未成交，原因客户不想要了",
chat_id=chat,
chat_type="group",
),
flow=self.flow,
ledger=self.ledger,
)
self.assertEqual(phase, "done")
ticket = self.ledger.get_ticket(work_order_no=wo)
self.assertEqual(ticket.status, "未成交")
self.assertEqual(ticket.lost_reason, "客户不想要了")
def test_sales_can_close_deal_by_text(self) -> None:
"""群里销售说成交，要入账并回话，不能装没听见。相近说法也认。"""
wo, chat = self._open_group()
phase = handle_group_collab(
InboundMessage(
sender_id="sales1",
message_id="deal-text-1",
content="成交",
chat_id=chat,
chat_type="group",
),
flow=self.flow,
ledger=self.ledger,
)
self.assertEqual(phase, "done")
ticket = self.ledger.get_ticket(work_order_no=wo)
self.assertEqual(ticket.status, "已成交")
texts = [x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text"]
self.assertTrue(any("已记录为「已成交」" in t for t in texts))
def test_product_cannot_close_deal_by_text(self) -> None:
wo, chat = self._open_group()
phase = handle_group_collab(
InboundMessage(
sender_id="prod4",
message_id="deal-text-2",
content="这个成了",
chat_id=chat,
chat_type="group",
),
flow=self.flow,
ledger=self.ledger,
)
self.assertEqual(phase, "not_sales")
ticket = self.ledger.get_ticket(work_order_no=wo)
self.assertNotEqual(ticket.status, "已成交")
texts = [x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text"]
self.assertTrue(any("只有负责该工单的销售" in t for t in texts))
def test_stale_deal_card_speaks(self) -> None:
from agent.policy.sea_group_ops import explain_deal_click
self._open_group()
sess = self.flow.session_of("sales1")
sess.deal_version = 2
sess.collab_chat_id = sess.collab_chat_id or "wr_sea_1"
reason = explain_deal_click(
flow=self.flow,
sender_id="sales1",
action=copy.BTN_DEAL,
card_meta={"event_key": f"deal:{sess.work_order_no}:v1"},
sess=sess,
)
self.assertEqual(reason, "stale_deal")
texts = [x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text"]
self.assertTrue(any("已过期" in t for t in texts))
def test_sales_who_is_also_product_can_quote(self) -> None:
"""拉群销售若后台也是产品·海运，群里按行改价必须收下，不能装没看见。"""
self.ledger.set_sea_staff(
[
{
"name": "吴吉林",
"wecomId": "sales1",
"roleCode": "sea",
"routes": "海运·全部线路",
"status": "active",
},
{
"name": "赵产品",
"wecomId": "prod4",
"roleCode": "sea",
"routes": "海运·全部线路",
"status": "active",
},
]
)
wo, chat = self._open_group()
ticket = self.ledger.get_ticket(work_order_no=wo)
self.assertIn("sales1", ticket.product_ids)
phase = handle_group_collab(
InboundMessage(
sender_id="sales1",
message_id="dual-role-quote",
content="码头操作费100CNY\n报关费改为300CNY",
chat_id=chat,
chat_type="group",
),
flow=self.flow,
ledger=self.ledger,
injected_intent="product_quote",
)
self.assertEqual(phase, "wait_tms_choice")
def test_existing_ticket_sales_later_product_role_can_quote(self) -> None:
"""已建群后销售才挂上产品岗：仍按后台岗位认，不能只看建群时的产品名单。"""
wo, chat = self._open_group()
ticket = self.ledger.get_ticket(work_order_no=wo)
self.assertNotIn("sales1", ticket.product_ids)
self.ledger.set_sea_staff(
[
{
"name": "吴吉林",
"wecomId": "sales1",
"roleCode": "sea",
"routes": "海运·全部线路",
"status": "active",
}
]
)
phase = handle_group_collab(
InboundMessage(
sender_id="sales1",
message_id="later-product-quote",
content="码头操作费100CNY\n报关费改为300CNY",
chat_id=chat,
chat_type="group",
),
flow=self.flow,
ledger=self.ledger,
injected_intent="product_quote",
)
self.assertEqual(phase, "wait_tms_choice")
def test_product_fee_line_quote_asks_then_keeps_tms(self) -> None:
"""有 TMS 价时先问；选「在基础上改」才出单，没点名的 TMS 费用保留。"""
wo, chat = self._open_group()
ticket = self.ledger.get_ticket(work_order_no=wo)
ticket.quote = {
"total": "2287.65",
"currency": "CNY",
"source_label": "TMS 标准报价",
"fee_lines": [
{"name": "海运费", "amount": "100.00", "currency": "USD"},
{"name": "操作费", "amount": "680.00", "currency": "CNY"},
{"name": "报关费", "amount": "250.00", "currency": "CNY"},
],
}
phase = handle_group_collab(
InboundMessage(
sender_id="prod4",
message_id="fee-line-1",
content="码头操作费100CNY\n报关费改为300CNY",
chat_id=chat,
chat_type="group",
),
flow=self.flow,
ledger=self.ledger,
injected_intent="product_quote",
)
self.assertEqual(phase, "wait_tms_choice")
texts = [x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text"]
self.assertTrue(any("在TMS基础上改" in t and "不用TMS报价" in t for t in texts))
self.assertFalse(any("没有识别到报价" in t for t in texts))
ticket = self.ledger.get_ticket(work_order_no=wo)
before = {str(x.get("name") or ""): x for x in (ticket.quote.get("fee_lines") or [])}
self.assertEqual(str(before["报关费"].get("amount") or ""), "250.00")
phase = handle_group_collab(
InboundMessage(
sender_id="prod4",
message_id="fee-line-2",
content="在TMS基础上改",
chat_id=chat,
chat_type="group",
),
flow=self.flow,
ledger=self.ledger,
)
self.assertEqual(phase, "product_quote_sent")
ticket = self.ledger.get_ticket(work_order_no=wo)
names = {
str(x.get("name") or ""): x
for x in (ticket.quote.get("fee_lines") or ticket.quote.get("fee_rows") or [])
}
self.assertEqual(str(names["报关费"].get("amount") or ""), "300")
self.assertEqual(str(names["操作费"].get("amount") or ""), "680.00")
self.assertEqual(str(names["码头操作费"].get("amount") or ""), "100")
self.assertEqual(str(names["海运费"].get("amount") or ""), "100.00")
def test_product_text_on_miss_is_manual_not_tms(self) -> None:
"""TMS 无价时产品文字报价：摘要 TMS 仍是暂无，人工报价两行齐全。"""
facts = dict(COMPLETE_SEA)
facts["目的港"] = "无价港"
drive_sea(self.flow,
sender_id="sales1",
text="海运无价",
reply=self._reply,
injected_facts=facts,
injected_mode="SEA",
)
self.flow.on_button(
sender_id="sales1",
action=copy.BTN_PULL_COLLAB,
reply=self._reply,
)
sess = self.flow.session_of("sales1")
chat = sess.collab_chat_id
phase = handle_group_collab(
InboundMessage(
sender_id="prod4",
message_id="miss-manual-1",
content="海运费：USD 100.00\n燃油附加费：USD 15.00",
chat_id=chat,
chat_type="group",
),
flow=self.flow,
ledger=self.ledger,
)
self.assertEqual(phase, "product_quote_sent")
briefs = [x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text"]
text = "\n".join(briefs)
self.assertIn("TMS 暂无匹配报价", text)
self.assertIn("人工报价：", text)
self.assertIn("海运费：USD 100.00", text)
self.assertIn("燃油附加费：USD 15.00", text)
self.assertNotIn("确认报价", text)
def test_confirm_tms_saves_sheet_and_notifies_app(self) -> None:
"""有 TMS 价时产品回确认报价：模板进后台，群里告诉大家已保存。"""
from agent.routing.group_intent import INTENT_CONFIRM_TMS, classify_group_text
self.assertEqual(classify_group_text("确认报价", allow_network=False), INTENT_CONFIRM_TMS)
wo, chat = self._open_group()
phase = handle_group_collab(
InboundMessage(
sender_id="prod4",
message_id="confirm-tms-1",
content="确认报价",
chat_id=chat,
chat_type="group",
),
flow=self.flow,
ledger=self.ledger,
)
self.assertEqual(phase, "quote_confirmed")
ticket = self.ledger.get_ticket(work_order_no=wo)
self.assertEqual(ticket.status, "已报价")
self.assertTrue(ticket.attachments)
saved_name = str(ticket.attachments[0].get("file_name") or "")
self.assertEqual(saved_name, f"{wo}海运默认.xlsx")
texts = [x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text"]
self.assertIn(f"工单{wo}的1个附件已保存到后台工单", texts)
self.assertTrue(any("成交" in t for t in texts))
def test_first_option_text_keeps_tms(self) -> None:
"""问句列了 1/2 时，回「第一个」要按在 TMS 基础上改，不能装没听见。"""
from agent.routing.group_intent import INTENT_KEEP_TMS, classify_group_text
self.assertEqual(classify_group_text("第一个", allow_network=False), INTENT_KEEP_TMS)
wo, chat = self._open_group()
ticket = self.ledger.get_ticket(work_order_no=wo)
ticket.quote = {
"total": "2287.65",
"currency": "CNY",
"source_label": "TMS 标准报价",
"fee_lines": [
{"name": "操作费", "amount": "680", "currency": "CNY", "unit_price": "680"},
{"name": "文件费", "amount": "450", "currency": "CNY", "unit_price": "450"},
],
}
handle_group_collab(
InboundMessage(
sender_id="prod4",
message_id="first-opt-1",
content="操作费改为700CNY / 文件费改为500CNY",
chat_id=chat,
chat_type="group",
),
flow=self.flow,
ledger=self.ledger,
injected_intent="product_quote",
)
phase = handle_group_collab(
InboundMessage(
sender_id="prod4",
message_id="first-opt-2",
content="第一个",
chat_id=chat,
chat_type="group",
),
flow=self.flow,
ledger=self.ledger,
)
self.assertEqual(phase, "product_quote_sent")
ticket = self.ledger.get_ticket(work_order_no=wo)
names = {
str(x.get("name") or ""): x
for x in (ticket.quote.get("fee_lines") or ticket.quote.get("fee_rows") or [])
}
self.assertEqual(str(names["操作费"].get("amount") or ""), "700")
self.assertEqual(str(names["文件费"].get("amount") or ""), "500")
def test_other_product_can_drop_tms(self) -> None:
"""任一产品都能选；说不用 TMS 就只留刚报的价。"""
self.ledger.set_sea_staff(
[
{
"name": "赵产品",
"wecomId": "prod4",
"roleCode": "sea",
"routes": "海运·全部线路",
"status": "active",
},
{
"name": "钱产品",
"wecomId": "prod5",
"roleCode": "sea",
"routes": "海运·全部线路",
"status": "active",
},
]
)
wo, chat = self._open_group()
ticket = self.ledger.get_ticket(work_order_no=wo)
ticket.quote = {
"source_label": "TMS 标准报价",
"fee_lines": [
{"name": "海运费", "amount": "100.00", "currency": "USD"},
{"name": "报关费", "amount": "250.00", "currency": "CNY"},
],
}
handle_group_collab(
InboundMessage(
sender_id="prod4",
message_id="drop-1",
content="报关费改为300CNY",
chat_id=chat,
chat_type="group",
),
flow=self.flow,
ledger=self.ledger,
injected_intent="product_quote",
)
phase = handle_group_collab(
InboundMessage(
sender_id="prod5",
message_id="drop-2",
content="不要TMS的价，按我刚才报的",
chat_id=chat,
chat_type="group",
),
flow=self.flow,
ledger=self.ledger,
)
self.assertEqual(phase, "product_quote_sent")
ticket = self.ledger.get_ticket(work_order_no=wo)
names = {
str(x.get("name") or ""): x
for x in (ticket.quote.get("fee_lines") or ticket.quote.get("fee_rows") or [])
}
self.assertEqual(list(names), ["报关费"])
self.assertEqual(str(names["报关费"].get("amount") or ""), "300")
def test_parse_group_file_fills_media(self) -> None:
from agent.channel.wecom.parse import parse_plain_xml
xml = (
"<xml><FromUserName>prod4</FromUserName><CreateTime>1</CreateTime>"
"<MsgType>file</MsgType><MediaId>mid-x</MediaId>"
"<FileName>报价.xlsx</FileName><ChatId>wr_sea_1</ChatId></xml>"
)
msg = parse_plain_xml(xml)
self.assertEqual(msg.chat_type, "group")
self.assertEqual(msg.chat_id, "wr_sea_1")
self.assertEqual(msg.media.get("filename"), "报价.xlsx")
self.assertEqual(msg.media.get("media_id"), "mid-x")
if __name__ == "__main__":
unittest.main()
Reference in New Issue View Git Blame Copy Permalink