170 lines
7.3 KiB
Python
170 lines
7.3 KiB
Python
"""
|
|
空运增值服务：多项原词、示例行不当已填、传给 TMS 的是 addedServices 列表。
|
|
"""
|
|
|
|
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")
|
|
|
|
from agent.llm.extract_text import extract_inquiry_snapshot
|
|
from agent.policy import inquiry_copy as copy
|
|
from agent.schema.added_services import harvest_added_services, split_added_services
|
|
from agent.schema.tms_air_query import assemble_air_query
|
|
|
|
|
|
_EXAMPLE = "(如：托盘、前置仓、带电、卡转、卡转低消、MUST GO、备装、分泡、温控、快件、DG、TC）"
|
|
|
|
|
|
class AirAddedServicesTests(unittest.TestCase):
|
|
def test_split_enumeration_and_chinese_comma(self) -> None:
|
|
"""顿号、中文逗号、英文逗号拆成同一组，空格不拆开 MUST GO。"""
|
|
self.assertEqual(
|
|
split_added_services("托盘、前置仓、带电"),
|
|
["托盘", "前置仓", "带电"],
|
|
)
|
|
self.assertEqual(
|
|
split_added_services("托盘，前置仓，带电"),
|
|
["托盘", "前置仓", "带电"],
|
|
)
|
|
self.assertEqual(
|
|
split_added_services("托盘,前置仓,带电"),
|
|
["托盘", "前置仓", "带电"],
|
|
)
|
|
self.assertEqual(split_added_services("MUST GO、带电"), ["MUST GO", "带电"])
|
|
|
|
def test_example_line_is_not_filled(self) -> None:
|
|
text = "增值服务（非必填）：\n" + _EXAMPLE
|
|
self.assertEqual(harvest_added_services(text), [])
|
|
|
|
def test_labeled_value_keeps_order_and_canonical_english(self) -> None:
|
|
dun = harvest_added_services("增值服务（非必填）：托盘、前置仓、带电")
|
|
comma = harvest_added_services("增值服务（非必填）：托盘，前置仓，带电")
|
|
self.assertEqual(dun, ["托盘", "前置仓", "带电"])
|
|
self.assertEqual(comma, ["托盘", "前置仓", "带电"])
|
|
spoken = harvest_added_services("这票 must go，要dg和tc")
|
|
self.assertEqual(spoken, ["MUST GO", "DG", "TC"])
|
|
|
|
def test_longest_match_and_negation(self) -> None:
|
|
self.assertEqual(harvest_added_services("要卡转低消"), ["卡转低消"])
|
|
self.assertNotIn("卡转", harvest_added_services("要卡转低消"))
|
|
self.assertEqual(harvest_added_services("不带电"), [])
|
|
self.assertEqual(harvest_added_services("带电"), ["带电"])
|
|
|
|
def test_outside_list_only_when_labeled(self) -> None:
|
|
"""名单外的词必须写在增值服务后面才记；12 个词只要出现就记。"""
|
|
self.assertEqual(harvest_added_services("要打木架"), [])
|
|
self.assertEqual(
|
|
harvest_added_services("增值服务：打木架"),
|
|
["打木架"],
|
|
)
|
|
self.assertEqual(
|
|
harvest_added_services("包装纸箱，要卡转低消，带电，增值服务：打木架"),
|
|
["卡转低消", "带电", "打木架"],
|
|
)
|
|
|
|
def test_extract_stores_added_services_and_keeps_other_package(self) -> None:
|
|
snap = extract_inquiry_snapshot(
|
|
"空运到吉隆坡，茶叶，10件，100公斤，2方，包装纸箱，要卡转低消，带电，增值服务：打木架",
|
|
injected_mode="AIR",
|
|
)
|
|
facts = snap["facts"]
|
|
self.assertEqual(facts.get("added_services"), "卡转低消、带电、打木架")
|
|
self.assertEqual(facts.get("包装方式"), "纸箱")
|
|
self.assertEqual(facts.get("是否含电"), "是")
|
|
|
|
def test_pallet_does_not_replace_another_package(self) -> None:
|
|
snap = extract_inquiry_snapshot(
|
|
"空运，包装是纸箱，要托盘",
|
|
injected_mode="AIR",
|
|
)
|
|
facts = snap["facts"]
|
|
self.assertEqual(facts.get("包装方式"), "纸箱")
|
|
self.assertEqual(facts.get("added_services"), "托盘")
|
|
|
|
def test_blank_service_line_does_not_copy_package_or_next_field(self) -> None:
|
|
"""包装类型写托盘、增值服务留空时，不能把托盘和下一行报价日期抄进增值服务。"""
|
|
text = (
|
|
"运输方式：空运\n"
|
|
"起运地：-\n"
|
|
"目的地：CRK\n"
|
|
"品名：LMM振动测试定位治具\n"
|
|
"件数：1件\n"
|
|
"重量（KG）：1.5KG\n"
|
|
"体积（CBM）：100CBM\n"
|
|
"包装类型：托盘\n"
|
|
"增值服务：\n"
|
|
"报价日期（非必填）："
|
|
)
|
|
self.assertEqual(harvest_added_services(text), [])
|
|
snap = extract_inquiry_snapshot(text)
|
|
self.assertNotIn("added_services", snap["facts"])
|
|
self.assertEqual(snap["facts"].get("包装方式"), "托盘")
|
|
card = copy.air_confirm_card(snap["facts"])
|
|
self.assertIn("包装类型（必填）：托盘", card)
|
|
self.assertIn("增值服务（非必填）：\n", card)
|
|
self.assertNotIn("托盘、报价日期", card)
|
|
|
|
def test_pallet_alone_is_package_and_service(self) -> None:
|
|
snap = extract_inquiry_snapshot(
|
|
"空运，不带电，包装托盘，目的地吉隆坡",
|
|
injected_mode="AIR",
|
|
)
|
|
facts = snap["facts"]
|
|
self.assertEqual(facts.get("包装方式"), "托盘")
|
|
self.assertEqual(facts.get("added_services"), "托盘")
|
|
self.assertNotIn("带电", facts.get("added_services") or "")
|
|
self.assertEqual(facts.get("是否含电"), "否")
|
|
|
|
def test_confirm_lists_optional_and_ticket_hides_empty(self) -> None:
|
|
card = copy.air_confirm_card({"目的港": "吉隆坡", "包装方式": "纸箱"})
|
|
self.assertIn("增值服务（非必填）：", card)
|
|
clarify = copy.ask_clarify(
|
|
facts={"目的港": "吉隆坡"},
|
|
missing_keys=["品名"],
|
|
transport_mode="空运",
|
|
)
|
|
self.assertIn("增值服务（非必填）：\n" + _EXAMPLE, clarify)
|
|
lines = copy.inquiry_field_lines(
|
|
work_order_no="WO1",
|
|
facts={"目的港": "吉隆坡", "包装方式": "纸箱", "added_services": "托盘、带电"},
|
|
transport_mode="空运",
|
|
)
|
|
self.assertIn("增值服务：托盘、带电", lines)
|
|
empty = copy.inquiry_field_lines(
|
|
work_order_no="WO1",
|
|
facts={"目的港": "吉隆坡", "包装方式": "纸箱"},
|
|
transport_mode="空运",
|
|
)
|
|
self.assertFalse(any(line.startswith("增值服务") for line in empty))
|
|
|
|
def test_assemble_keeps_services_without_blocking(self) -> None:
|
|
ready = {
|
|
"目的港": "吉隆坡",
|
|
"品名": "茶叶",
|
|
"件数": "10",
|
|
"毛重": "100",
|
|
"体积": "2",
|
|
"包装方式": "纸箱",
|
|
"报价日期": "2026-10-13",
|
|
"added_services": "托盘、前置仓、带电",
|
|
}
|
|
out = assemble_air_query(ready)
|
|
self.assertTrue(out["ok"], out)
|
|
self.assertEqual(out["facts"]["added_services"], "托盘、前置仓、带电")
|
|
bare = dict(ready)
|
|
bare.pop("added_services")
|
|
self.assertTrue(assemble_air_query(bare)["ok"])
|
|
self.assertNotIn("added_services", assemble_air_query(bare)["facts"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|