看板平均响应按协同分发统计;查价失败可从主账恢复;报价模板匹配与填表更稳,后台改名运营后台。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
linmianxun
2026-10-05 18:00:08 +08:00
co-authored by Cursor
parent e95e8869a6
commit 58df358577
28 changed files with 1493 additions and 245 deletions
@@ -143,6 +143,61 @@ class QuoteTemplateMatchTests(unittest.TestCase):
self.assertIsNotNone(hit)
self.assertEqual(hit.name, "跨境整车")
def test_land_admin_domestic_ftl_keyword_does_not_steal_cross(self) -> None:
"""后台「国内整车」不得靠「整车」二字抢走跨境整车(WO202610030032)。"""
hit = pick_quote_template(
templates=[
{
"templateId": "dom",
"name": "国内陆运整车报价单",
"bizType": "陆运",
"keywords": ["国内整车"],
"isDefault": False,
},
{
"templateId": "xb",
"name": "跨境整车报价单",
"bizType": "陆运",
"keywords": ["东南亚"],
"isDefault": False,
},
],
business_line="LAND",
facts={
"运输类型": "跨境整车",
"线路类别": "东南亚",
"运输分类": "普件(东南亚、中亚、中欧涉及国内+国外段)",
"起运港": "上海",
"目的港": "春疆货场",
},
)
self.assertIsNotNone(hit)
self.assertEqual(hit.name, "跨境整车报价单")
def test_land_admin_domestic_ftl_still_hits_domestic_type(self) -> None:
hit = pick_quote_template(
templates=[
{
"templateId": "dom",
"name": "国内陆运整车报价单",
"bizType": "陆运",
"keywords": ["国内整车"],
"isDefault": False,
},
{
"templateId": "xb",
"name": "跨境整车报价单",
"bizType": "陆运",
"keywords": ["东南亚"],
"isDefault": False,
},
],
business_line="LAND",
facts={"运输类型": "国内运输整车", "线路类别": "国内长途/零担"},
)
self.assertIsNotNone(hit)
self.assertEqual(hit.name, "国内陆运整车报价单")
def test_land_falls_back_to_default(self) -> None:
hit = pick_quote_template(
templates=[
@@ -22,8 +22,11 @@ from agent.policy.system_exception import (
SystemExceptionEvent,
call_with_retries,
confirm_system_exception,
confirm_tms_quote_if_tech,
extract_work_order_nos,
is_tms_tech_failure,
query_tms_resilient,
recover_tms_query_from_ledger,
should_mark_ticket,
ticket_is_paused,
)
@@ -196,6 +199,89 @@ class TestSystemExceptionRules(unittest.TestCase):
self.assertEqual(out.get("classification"), "TMS_TECH_FAILURE")
class TestTmsTimeoutRecoverPriced(unittest.TestCase):
"""查到价但智能体超时:主账已有价则按成功续办,不标 TMS异常。"""
def setUp(self) -> None:
self.ledger = MemoryLedger()
created = self.ledger.create_ticket(
sender_id="sales1",
business_line="AIR",
facts={"起运港": "南京", "目的港": "吉隆坡", "品名": "茶叶"},
)
self.wo = str(created.get("work_order_no") or "")
self.ledger.upsert_quote(
work_order_no=self.wo,
quote={
"total": "CNY 5001.00",
"currency": "CNY",
"source": "TMS",
"source_label": "TMS标准报价",
},
to_status="已报价",
)
def test_recover_from_ledger_when_already_quoted(self) -> None:
got = recover_tms_query_from_ledger(self.ledger, work_order_no=self.wo)
self.assertIsNotNone(got)
assert got is not None
self.assertTrue(got.get("ok"))
self.assertTrue(got.get("has_price"))
self.assertTrue(got.get("recovered"))
self.assertIn("5001", str((got.get("quote") or {}).get("total") or ""))
def test_query_resilient_recovers_after_tech_fail(self) -> None:
self.ledger.query_tms = ( # type: ignore[method-assign]
lambda **_kwargs: {
"ok": False,
"classification": "TMS_TECH_FAILURE",
"has_price": False,
"quote": {},
}
)
out = query_tms_resilient(
self.ledger, work_order_no=self.wo, facts={"起运港": "南京", "目的港": "吉隆坡"}
)
self.assertTrue(out.get("ok"))
self.assertTrue(out.get("has_price"))
self.assertTrue(out.get("recovered"))
self.assertEqual(out.get("classification"), "HAS_PRICE")
def test_confirm_skips_mark_when_ledger_priced(self) -> None:
replies: list[str] = []
marked = confirm_tms_quote_if_tech(
ledger=self.ledger,
reply=replies.append,
work_order_no=self.wo,
tms={"ok": False, "classification": "TMS_TECH_FAILURE"},
sender_id="sales1",
ticket_status="询价中",
)
self.assertFalse(marked)
self.assertEqual(replies, [])
snap = self.ledger.get_for_agent(work_order_no=self.wo, sender_id="sales1")
self.assertFalse(str(snap.get("system_exception") or "").strip())
def test_true_tech_fail_still_marks_when_no_price(self) -> None:
created = self.ledger.create_ticket(
sender_id="sales2",
business_line="AIR",
facts={"起运港": "南京", "目的港": "吉隆坡", "品名": "茶叶"},
)
wo2 = str(created.get("work_order_no") or "")
replies: list[str] = []
marked = confirm_tms_quote_if_tech(
ledger=self.ledger,
reply=replies.append,
work_order_no=wo2,
tms={"ok": False, "classification": "TMS_TECH_FAILURE"},
sender_id="sales2",
ticket_status="询价中",
)
self.assertTrue(marked)
self.assertTrue(any("TMS接口异常" in x for x in replies))
class TestConfirmSystemException(unittest.TestCase):
def setUp(self) -> None:
self.ledger = MemoryLedger()