""" 陆运群内协同：摘要只列本单协同项、补字段过滤、预填。 不连企微 / 真实 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.ledger.memory_ledger import MemoryLedger from agent.policy import inquiry_copy as copy from agent.policy.land_text_flow import LandTextInquiryFlow from agent.routing.group_intent import INTENT_ADJUST, INTENT_FIELDS, INTENT_QUOTE class CountingLedger(MemoryLedger): def __init__(self) -> None: super().__init__() self.last_created_no = "" def create_ticket(self, **kwargs): out = super().create_ticket(**kwargs) self.last_created_no = str(out.get("work_order_no") or "") return out class FakeGroupClient: def __init__(self) -> None: self.created: list[dict] = [] self.sent: list[dict] = [] def create_group(self, *, name: str, userids: list[str]) -> dict: self.created.append({"name": name, "userids": list(userids)}) return {"ok": True, "chat_id": "wr_land_collab", "error": ""} def add_group_members(self, *, chat_id: str, userids: list[str]) -> dict: 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"}) 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"}) return {"ok": True} class LandGroupCollabTests(unittest.TestCase): def setUp(self) -> None: self.ledger = CountingLedger() self.ledger.set_quote_templates( [ { "templateId": "mem-default-land", "name": "陆运默认", "bizType": "陆运", "keywords": [], "isDefault": True, } ] ) self.ledger.set_land_staff( [ { "roleCode": "land", "status": "active", "wecomId": "p_dom", "name": "乙", "routes": "国内长途/零担,国内城配/拖车/打包,中港/中亚/中欧,东南亚", } ] ) self.groups = FakeGroupClient() self.flow = LandTextInquiryFlow(self.ledger, group_client=self.groups) self.replies: list[str] = [] def _reply(self, text: str, extra=None) -> str: self.replies.append(text) return "" def _group_texts(self) -> str: return "\n".join(x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text") def _open( self, *, option_text: str, extra_facts: dict | None = None, cargo: dict | None = None, ) -> tuple[str, str]: facts = { "起运港": "广州", "目的港": "深圳", "品名": "衣服", "体积": "100CBM", "毛重": "50kg", } facts.update(cargo or {}) facts.update(extra_facts or {}) self.flow.on_text( sender_id="u1", text=option_text, reply=self._reply, injected_facts=facts, injected_mode="LAND", ) self.flow.on_text(sender_id="u1", text="确定", reply=self._reply) self.replies.clear() phase = self.flow.on_button( sender_id="u1", action=copy.BTN_PULL_COLLAB, reply=self._reply, ) self.assertEqual(phase, "collab_group") wo = self.flow.session_of("u1").work_order_no return wo, "wr_land_collab" def _msg(self, *, sender: str, text: str, mid: str) -> InboundMessage: return InboundMessage( sender_id=sender, message_id=mid, content=text, msg_type="text", chat_id="wr_land_collab", chat_type="group", ) def test_ltl_brief_only_four_collab_and_prefill_pack(self) -> None: self._open( option_text="国内运输拼车 国内长途/零担 急件 包装方式纸箱", extra_facts={"包装方式": "纸箱"}, ) brief = self._group_texts() self.assertIn("包装方式：纸箱", brief) self.assertIn("客户名称：-", brief) self.assertIn("货值：-", brief) self.assertIn("是否为危险品：-", brief) self.assertNotIn("贸易条款", brief) self.assertNotIn("商品海关编码", brief) self.assertNotIn("海关编码", brief) def test_domestic_ftl_longhaul_can_pull_and_brief_order(self) -> None: """国内整车配长途：允许拉群；协同顺序是客户名称、货值、包装方式、是否为危险品。""" self._open( option_text="国内运输整车 国内长途/零担 急件", cargo={"车型/数量": "9米/1"}, ) self.assertTrue(self.groups.created) brief = self._group_texts() collab_part = brief.split("协同补充字段")[1] self.assertIn("客户名称：-", collab_part) self.assertIn("货值：-", collab_part) self.assertIn("包装方式：-", collab_part) self.assertIn("是否为危险品：-", collab_part) self.assertLess(collab_part.find("客户名称"), collab_part.find("货值")) self.assertLess(collab_part.find("货值"), collab_part.find("包装方式")) self.assertLess(collab_part.find("包装方式"), collab_part.find("是否为危险品")) self.assertNotIn("贸易条款", brief) self.assertNotIn("海关编码", brief) # 车型/数量只在询价字段，不进协同。 self.assertIn("车型/数量：9米/1", brief.split("协同补充字段")[0]) self.assertNotIn("车型/数量", collab_part) def test_cross_ftl_vehicle_stays_inquiry(self) -> None: wo, _ = self._open( option_text="跨境整车 东南亚 急件", cargo={ "目的港": "河内", "车型/数量": "9米/1", }, ) brief = self._group_texts() self.assertIn("车型/数量：9米/1", brief.split("协同补充字段")[0]) self.assertNotIn("车型/数量", brief.split("协同补充字段")[1]) self.groups.sent.clear() phase = handle_group_collab( self._msg(sender="u1", text="车型/数量 12米/1", mid="v1"), flow=self.flow, ledger=self.ledger, injected_intent=INTENT_FIELDS, ) self.assertEqual(phase, "collab_not_recognized") ticket = self.ledger.get_ticket(work_order_no=wo) self.assertEqual(ticket.facts.get("车型/数量"), "9米/1") self.assertFalse(ticket.collab_facts.get("车型/数量")) def test_hk_ftl_vehicle_is_collab(self) -> None: wo, _ = self._open( option_text="中港整车 中港/中亚/中欧 急件", cargo={"目的港": "香港"}, ) brief = self._group_texts() collab_part = brief.split("协同补充字段")[1] self.assertIn("车型/数量：-", collab_part) self.assertNotIn("车型/数量", brief.split("协同补充字段")[0]) self.groups.sent.clear() phase = handle_group_collab( self._msg(sender="u1", text="车型/数量 9米/1", mid="hk1"), flow=self.flow, ledger=self.ledger, injected_intent=INTENT_FIELDS, ) self.assertEqual(phase, "collab_field_saved") ticket = self.ledger.get_ticket(work_order_no=wo) self.assertEqual(ticket.collab_facts.get("车型/数量"), "9米/1") self.assertFalse(ticket.facts.get("车型/数量")) self.assertIn("车型/数量：9米/1", self._group_texts()) def test_ltl_rejects_trade_terms_only(self) -> None: self._open(option_text="国内运输拼车 国内长途/零担 急件") self.groups.sent.clear() phase = handle_group_collab( self._msg(sender="u1", text="贸易条款 FOB", mid="t1"), flow=self.flow, ledger=self.ledger, injected_intent=INTENT_FIELDS, ) self.assertEqual(phase, "collab_not_recognized") text = self._group_texts() self.assertIn("客户名称", text) self.assertNotIn("货好时间", text) def test_ltl_keeps_name_drops_trade_terms(self) -> None: wo, _ = self._open(option_text="国内运输拼车 国内长途/零担 急件") self.groups.sent.clear() phase = handle_group_collab( self._msg(sender="u1", text="客户名称张三，贸易条款 FOB", mid="t2"), flow=self.flow, ledger=self.ledger, injected_intent=INTENT_FIELDS, ) self.assertEqual(phase, "collab_field_saved") ticket = self.ledger.get_ticket(work_order_no=wo) self.assertEqual(ticket.collab_facts.get("客户名称"), "张三") self.assertFalse(ticket.collab_facts.get("贸易条款")) self.assertIn("客户名称：张三", self._group_texts()) def test_dg_and_value_and_bare_pack(self) -> None: wo, _ = self._open(option_text="国内运输拼车 国内长途/零担 急件") handle_group_collab( self._msg(sender="u1", text="普货", mid="dg1"), flow=self.flow, ledger=self.ledger, injected_intent=INTENT_FIELDS, ) ticket = self.ledger.get_ticket(work_order_no=wo) self.assertEqual(ticket.collab_facts.get("是否为危险品"), "否") handle_group_collab( self._msg(sender="u1", text="危险品", mid="dg2"), flow=self.flow, ledger=self.ledger, injected_intent=INTENT_FIELDS, ) ticket = self.ledger.get_ticket(work_order_no=wo) self.assertEqual(ticket.collab_facts.get("是否为危险品"), "是") handle_group_collab( self._msg(sender="u1", text="货值 10万 包装方式编织袋", mid="v1"), flow=self.flow, ledger=self.ledger, injected_intent=INTENT_FIELDS, ) ticket = self.ledger.get_ticket(work_order_no=wo) self.assertEqual(ticket.collab_facts.get("货值"), "10万") self.assertEqual(ticket.collab_facts.get("包装方式"), "编织袋") self.groups.sent.clear() phase = handle_group_collab( self._msg(sender="u1", text="编织袋", mid="v2"), flow=self.flow, ledger=self.ledger, injected_intent=INTENT_FIELDS, ) self.assertEqual(phase, "collab_not_recognized") def test_product_can_quote_without_collab(self) -> None: wo, _ = self._open(option_text="国内运输拼车 国内长途/零担 急件") phase = handle_group_collab( self._msg(sender="p_dom", text="运费 1000", mid="q1"), flow=self.flow, ledger=self.ledger, injected_intent=INTENT_QUOTE, injected_quote={"total": "1000", "source_label": "产品报价"}, ) self.assertIn(phase, {"product_quote_sent", "wait_tms_choice"}) ticket = self.ledger.get_ticket(work_order_no=wo) self.assertTrue(ticket.quote) def test_sales_adjust_sends_excel_and_deal(self) -> None: wo, _ = self._open(option_text="国内运输拼车 国内长途/零担 急件") self.groups.sent.clear() phase = handle_group_collab( self._msg(sender="u1", text="空运费加 10%", mid="adj1"), flow=self.flow, ledger=self.ledger, injected_intent=INTENT_ADJUST, ) self.assertEqual(phase, "group_adjusted") texts = self._group_texts() self.assertIn("已识别调价指令", texts) self.assertIn("调价后价格", texts) self.assertTrue(any(x.get("kind") == "file" for x in self.groups.sent)) ticket = self.ledger.get_ticket(work_order_no=wo) self.assertEqual(ticket.status, "已报价") def test_product_cannot_adjust(self) -> None: self._open(option_text="国内运输拼车 国内长途/零担 急件") self.groups.sent.clear() phase = handle_group_collab( self._msg(sender="p_dom", text="空运费减 5%", mid="adj2"), flow=self.flow, ledger=self.ledger, injected_intent=INTENT_ADJUST, ) self.assertEqual(phase, "adjust_not_sales") self.assertIn("只有这张单的销售能调价", self._group_texts()) if __name__ == "__main__": unittest.main()