""" 陆运核对卡与无产品话术。 """ 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()