Files

346 lines
13 KiB
Python

"""
陆运核对卡与无产品话术。
"""
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)
from agent.policy import inquiry_copy as copy
class LandCopyTest(unittest.TestCase):
def test_domestic_card_no_customs_no_wo(self):
text = copy.land_confirm_card(
{
"运输类型": "国内运输拼车",
"线路类别": "国内长途/零担",
"运输分类": "急件",
"起运港": "广州",
"目的港": "深圳",
"品名": "衣服",
"体积": "100CBM",
"毛重": "50kg",
"报价日期": "2026-09-17",
}
)
self.assertIn("始发站(必填):广州", text)
self.assertNotIn("通关口岸", text)
self.assertNotIn("WO", text)
self.assertIn('请回复“确定”进行TMS查价,内容有误时,请重新发内容。', text)
self.assertIn("运输类型", text)
self.assertIn("国内运输拼车", text)
def test_cross_empty_customs(self):
text = copy.land_confirm_card(
{
"运输类型": "跨境集拼",
"线路类别": "东南亚",
"运输分类": "急件",
"起运港": "广州",
"目的港": "河内",
"品名": "衣服",
"体积": "100CBM",
"毛重": "50kg",
"报价日期": "2026-09-17",
}
)
self.assertIn("通关口岸(非必填):", text)
def test_hk_lcl_shows_vehicle_not_load_field(self):
text = copy.land_confirm_card(
{
"运输类型": "中港零担/集拼",
"线路类别": "东南亚",
"运输分类": "急件",
"起运港": "广州",
"目的港": "香港",
"品名": "衣服",
"车型/数量": "9米6/1",
"报价日期": "2026-09-17",
}
)
self.assertIn("运输类型(必填):中港零担/集拼", text)
self.assertIn("车型/数量(必填):9米6/1", text)
self.assertNotIn("中港装载方式", text)
self.assertIn("通关口岸(非必填)", text)
def test_hk_full_truck_hides_vehicle(self):
text = copy.land_confirm_card(
{
"运输类型": "中港整车",
"线路类别": "东南亚",
"运输分类": "急件",
"起运港": "广州",
"目的港": "香港",
"品名": "衣服",
"报价日期": "2026-09-17",
}
)
self.assertIn("运输类型(必填):中港整车", text)
self.assertNotIn("车型/数量", text)
self.assertNotIn("中港装载方式", text)
def test_domestic_ftl_one_field(self):
text = copy.land_confirm_card(
{
"运输类型": "国内运输整车",
"线路类别": "国内长途/零担",
"运输分类": "急件",
"起运港": "广州",
"目的港": "深圳",
"品名": "衣服",
"车型/数量": "9米6/1",
"报价日期": "2026-09-17",
}
)
self.assertIn("车型/数量(必填):9米6/1", text)
self.assertNotIn("\n数量(必填)", text)
def test_no_staff_copy(self):
self.assertIn("陆运产品岗", copy.land_group_no_staff())
def test_ask_land_fields_marks_required_and_lists_optional(self):
text = copy.ask_land_fields(
facts={
"运输类型": "跨境集拼",
"线路类别": "东南亚",
"运输分类": "急件",
"起运港": "广州",
"目的港": "河内",
"品名": "衣服",
"毛重": "50kg",
},
missing_keys=["体积"],
land_subtype="跨境集拼",
)
self.assertIn("始发站(必填):广州", text)
self.assertIn("待补充:", text)
self.assertIn("体积(必填)", text)
self.assertIn("通关口岸(非必填)", text)
self.assertNotIn(copy.LAND_CONFIRM_TAIL, text)
def test_ask_land_fields_domestic_has_no_customs(self):
text = copy.ask_land_fields(
facts={
"运输类型": "国内运输拼车",
"起运港": "广州",
"目的港": "深圳",
"品名": "衣服",
"毛重": "50kg",
},
missing_keys=["体积"],
land_subtype="国内运输拼车",
)
self.assertIn("体积(必填)", text)
self.assertNotIn("通关口岸", text)
def test_ask_land_fields_filled_customs_not_pending(self):
text = copy.ask_land_fields(
facts={
"运输类型": "中港整车",
"起运港": "广州",
"目的港": "香港",
"品名": "衣服",
"通关口岸": "皇岗",
},
missing_keys=["起运港"],
land_subtype="中港整车",
)
self.assertIn("通关口岸(非必填):皇岗", text)
pending = text.split("待补充:", 1)[1]
self.assertNotIn("通关口岸", pending)
def test_inquiry_card_lists_land_fields_and_wo(self):
text = copy.inquiry_card(
work_order_no="WO202609170006",
facts={
"运输类型": "国内运输拼车",
"线路类别": "国内长途/零担",
"运输分类": "急件",
"起运港": "广州",
"目的港": "深圳",
"品名": "衣服",
"体积": "100CBM",
"毛重": "50kg",
"报价日期": "2026-09-17",
},
transport_mode="陆运",
)
self.assertIn("工单号:WO202609170006", text)
self.assertIn("运输方式:陆运", text)
self.assertIn("运输类型:国内运输拼车", text)
self.assertIn("始发站:广州", text)
self.assertIn("货物品名:衣服", text)
self.assertIn("工单已创建,正在查询TMS报价。", text)
self.assertNotIn("件数", text)
def test_scheme_card_uses_land_fields_not_sea(self):
facts = {
"运输类型": "国内运输拼车",
"线路类别": "国内城配/拖车/打包",
"运输分类": "普件(东南亚、中港常用线路/常规车型)",
"起运港": "泉州市",
"目的港": "南京市",
"品名": "通信机柜及线缆配件",
"体积": "10CBM",
"毛重": "10KG",
"报价日期": "2026-09-17",
}
quote = {"total": "CNY 1200", "eta": "3天", "valid_until": "2026-09-20"}
text = copy.scheme_card(
work_order_no="WO202609170011",
status="已报价",
facts=facts,
quote=quote,
transport_mode="陆运",
)
self.assertIn("运输方式:陆运", text)
self.assertIn("运输类型:国内运输拼车", text)
self.assertIn("始发站:泉州市", text)
self.assertIn("货物品名:通信机柜及线缆配件", text)
self.assertIn("体积:10CBM", text)
self.assertNotIn("整柜或拼柜", text)
self.assertNotIn("箱型箱量", text)
self.assertNotIn("件数:", text)
self.assertNotIn("包装类型:", text)
rows = {
x["keyname"]: x["value"]
for x in copy.scheme_horizontal_rows(
work_order_no="WO202609170011",
facts=facts,
quote=quote,
transport_mode="陆运",
)
}
self.assertIn("陆运 泉州市→南京市", rows["线路"])
self.assertIn("国内运输拼车", rows["类型"])
self.assertIn("国内城配/拖车/打包", rows["类型"])
self.assertIn("通信机柜及线缆配件", rows["货物"])
self.assertIn("10CBM", rows["货物"])
self.assertNotIn("航线", rows)
self.assertNotIn("箱型", rows)
self.assertNotIn("包装", rows)
self.assertNotIn("空运", "".join(rows.values()))
self.assertNotIn("海运", "".join(rows.values()))
def test_scheme_card_ftl_shows_vehicle_combo(self):
facts = {
"运输类型": "国内运输整车",
"线路类别": "国内长途/零担",
"运输分类": "急件",
"起运港": "广州",
"目的港": "深圳",
"品名": "衣服",
"车型/数量": "9米6/1",
}
rows = {
x["keyname"]: x["value"]
for x in copy.scheme_horizontal_rows(
work_order_no="WO202609170012",
facts=facts,
quote={},
transport_mode="陆运",
)
}
self.assertIn("9米6/1", rows["货物"])
self.assertNotIn("CBM", rows["货物"])
def test_land_brief_only_ltl_collab(self):
text = copy.sea_group_brief(
work_order_no="WO1",
facts={
"运输类型": "国内运输拼车",
"线路类别": "国内长途/零担",
"运输分类": "急件",
"起运港": "广州",
"目的港": "深圳",
"品名": "衣服",
"体积": "1",
"毛重": "1",
"报价日期": "2026-09-18",
},
quote=None,
collab_facts={"包装方式": "纸箱"},
sales_name="销",
product_names=["产"],
has_price=False,
transport_mode="陆运",
)
self.assertIn("包装方式:纸箱", text)
self.assertIn("客户名称:-", text)
self.assertIn("货值:-", text)
self.assertIn("是否为危险品:-", text)
self.assertNotIn("贸易条款", text)
self.assertNotIn("商品海关编码", text)
self.assertNotIn("海关编码", text)
def test_land_brief_ftl_longhaul_order(self):
text = copy.sea_group_brief(
work_order_no="WO2",
facts={
"运输类型": "国内运输整车",
"线路类别": "国内长途/零担",
"运输分类": "急件",
"起运港": "广州",
"目的港": "深圳",
"品名": "衣服",
"车型/数量": "9米/1",
"报价日期": "2026-09-18",
},
quote=None,
collab_facts={},
sales_name="销",
product_names=["产"],
has_price=False,
transport_mode="陆运",
)
collab = text.split("协同补充字段")[1]
self.assertLess(collab.find("客户名称"), collab.find("货值"))
self.assertLess(collab.find("货值"), collab.find("包装方式"))
self.assertNotIn("贸易条款", text)
def test_land_unrecognized_lists_keys(self):
keys = ("客户名称", "包装方式", "货值", "是否为危险品")
text = copy.group_fields_not_recognized(keys=keys, business_line="LAND")
self.assertIn("客户名称", text)
self.assertNotIn("货好时间", text)
def test_image_copy(self):
self.assertEqual(copy.IMAGE_RECOGNIZING, "正在识别图片")
self.assertIn("没看清", copy.IMAGE_UNREADABLE)
self.assertIn("暂时没认出来", copy.IMAGE_TECH_FAIL)
merged = copy.image_fail_then_clarify(copy.IMAGE_UNREADABLE, copy.ASK_TRANSPORT)
self.assertTrue(merged.startswith(copy.IMAGE_UNREADABLE))
self.assertIn(copy.ASK_TRANSPORT, merged)
def test_attachment_copy_and_card(self):
self.assertEqual(copy.ATTACH_RECOGNIZING, "正在识别附件")
self.assertIn("没看清", copy.ATTACH_UNREADABLE)
self.assertIn("暂时没认出来", copy.ATTACH_TECH_FAIL)
payload = copy.attachment_upload_card("https://ai.ytd-scm.com/inquiry-form/upload/tok")
card = payload["template_card"]
self.assertEqual(payload["msgtype"], "template_card")
# 现网上传卡是 text_notice。news_notice 必须带 card_image,企微会 42044 后退成纯文本。
self.assertEqual(card["card_type"], "text_notice")
self.assertNotIn("card_image", card)
self.assertEqual(card["source"]["desc"], "附件询价")
self.assertEqual(card["main_title"]["title"], "上传询价附件")
self.assertIn("Excel", card["main_title"]["desc"])
self.assertIn("图片请直接在企微聊天窗口发送", card["main_title"]["desc"])
self.assertIn("图片请直接在当前企微会话发送", card["sub_title_text"])
self.assertEqual(card["jump_list"][0]["title"], "上传询价附件")
self.assertIn("/upload/tok", card["card_action"]["url"])
self.assertEqual(card["card_action"]["type"], 1)
self.assertNotIn("userid", card["card_action"]["url"])
if __name__ == "__main__":
unittest.main()