""" 海运文字询价主路径单测（不连企微 / 真实 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.handlers.text_inquiry import handle_text_inquiry from agent.ledger.memory_ledger import MemoryLedger from agent.policy import inquiry_copy as copy from agent.policy.air_text_flow import FlowSession from agent.policy.sea_text_flow import SeaTextInquiryFlow, reset_sea_text_flow_for_test from agent.schema.field_validate import SEA_COLLAB_FIELDS, validate_required_fields from agent.schema.tms_sea_query import assemble_sea_query COMPLETE_SEA = { "起运港": "上海", "目的港": "洛杉矶", "品名": "普货", "货量": "20吨", "整柜或拼柜": "整柜", "箱型箱量": "1x40HQ", "贸易条款": "FOB", "运输分类": "Port to Port", } class SeaCollabContractTests(unittest.TestCase): def test_collab_no_longer_asks_trade_terms(self): self.assertNotIn("贸易条款", SEA_COLLAB_FIELDS) self.assertEqual( SEA_COLLAB_FIELDS, ("货好时间", "HS编码", "是否含油", "是否含电", "是否含磁"), ) class FakeGroupClient: """单测假企微建群：不打外网。""" def __init__(self) -> None: self.created: list[dict] = [] self.sent: list[dict] = [] self.fail = False self.next_chat_id = "wr_sea_1" def create_group(self, *, name: str, userids: list[str]) -> dict: if self.fail: return {"ok": False, "error": "wecom_create_fail", "chat_id": ""} self.created.append({"name": name, "userids": list(userids)}) return {"ok": True, "chat_id": self.next_chat_id, "error": ""} def add_group_members(self, *, chat_id: str, userids: list[str]) -> dict: self.created.append({"name": "", "userids": list(userids), "chat_id": chat_id, "add": True}) 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", "mention_userids": list(mention_userids or []), } ) 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", "file_b64": file_b64} ) return {"ok": True} def send_group_card(self, *, chat_id: str, template_card: dict) -> dict: self.sent.append({"chat_id": chat_id, "kind": "card", "template_card": template_card}) return {"ok": True} def send_text(self, *, touser: str, content: str) -> dict: self.sent.append({"touser": touser, "content": content, "kind": "app_text"}) return {"ok": True} def dismiss_group(self, *, chat_id: str, userids: list) -> dict: self.sent.append({"chat_id": chat_id, "kind": "dismiss", "userids": list(userids)}) return {"ok": True} class SeaTextInquiryTests(unittest.TestCase): def setUp(self) -> None: self.ledger = MemoryLedger() self.ledger.set_quote_templates( [ { "templateId": "mem-default-sea", "name": "海运默认", "bizType": "海运", "keywords": [], "isDefault": True, } ] ) self.groups = FakeGroupClient() self.flow = SeaTextInquiryFlow(self.ledger, group_client=self.groups) self.replies: list[str] = [] self.extras: list[dict] = [] def _reply(self, text: str, extra=None) -> None: self.replies.append(text) self.extras.append(dict(extra) if extra else {}) def test_sea_pieces_fill_cargo_qty(self) -> None: facts = { "起运港": "上海", "目的港": "洛杉矶", "品名": "普货", "件数": "15件", "整柜或拼柜": "整柜", "箱型箱量": "1x40HQ", "贸易条款": "FOB", "运输分类": "Port to Port", } check = validate_required_fields(facts=facts, business_line="SEA") self.assertTrue(check["ok"]) self.assertEqual(check["facts"]["货量"], "15件") self.assertNotIn("货量", check["missing"]) def test_container_shaped_cargo_qty_uses_pieces(self) -> None: """货量被抽成箱型时，改用件数。20件、15件同一条规则。""" facts = { "起运港": "广州港", "目的港": "巴生港", "品名": "机器设备", "货量": "40HQ*1", "件数": "20件", "整柜或拼柜": "整柜", "箱型箱量": "40HQ*1", "贸易条款": "DOOR-DOOR", "运输分类": "Port to Port", } check = validate_required_fields(facts=facts, business_line="SEA") self.assertEqual(check["facts"]["货量"], "20件") self.assertEqual(check["facts"]["箱型箱量"], "40HQ*1") def test_sea_missing_container_only_asks_container(self) -> None: facts = { "起运港": "上海", "目的港": "洛杉矶", "品名": "茶叶", "货量": "20吨", "整柜或拼柜": "整柜", "贸易条款": "FOB", "运输分类": "Port to Port", } check = validate_required_fields(facts=facts, business_line="SEA") self.assertFalse(check["ok"]) self.assertEqual(check["missing"], ["箱型箱量"]) self.assertIn("箱型箱量", check["missing_display"]) self.assertNotIn("件数", check["missing"]) self.assertNotIn("毛重", check["missing"]) text = copy.ask_clarify( facts=check["facts"], missing_keys=check["missing"], transport_mode="海运", ) self.assertIn("运输方式：海运", text) self.assertIn("起运港：上海", text) self.assertIn("货物数量：20吨", text) self.assertIn("1.箱型箱量（必填）：\n(如：1x40HQ）", text) self.assertNotIn("箱型箱量：(如", text) self.assertNotIn("件数", text) self.assertNotIn("重量", text) self.assertNotIn("报价日期：(如", text) self.assertIn("报价日期：", text) def test_sea_requires_trade_terms_and_class(self) -> None: facts = { "起运港": "上海", "目的港": "洛杉矶", "品名": "普货", "货量": "20吨", "整柜或拼柜": "整柜", "箱型箱量": "1x40HQ", } check = validate_required_fields(facts=facts, business_line="SEA") self.assertFalse(check["ok"]) self.assertIn("贸易条款", check["missing"]) self.assertIn("运输分类", check["missing"]) self.assertIn("贸易条款", check["missing_display"]) self.assertIn("运输分类", check["missing_display"]) def test_sea_rejects_door_to_door_class(self) -> None: facts = { "起运港": "上海", "目的港": "洛杉矶", "品名": "普货", "货量": "20吨", "整柜或拼柜": "整柜", "箱型箱量": "1x40HQ", "贸易条款": "FOB", "运输分类": "门到门", } check = validate_required_fields(facts=facts, business_line="SEA") self.assertFalse(check["ok"]) self.assertIn("运输分类", check["missing"]) def test_sea_canonicalizes_p2p(self) -> None: facts = { "起运港": "上海", "目的港": "洛杉矶", "品名": "普货", "货量": "20吨", "整柜或拼柜": "整柜", "箱型箱量": "1x40HQ", "贸易条款": "到岸", "运输分类": "p2p", } check = validate_required_fields(facts=facts, business_line="SEA") self.assertTrue(check["ok"]) self.assertEqual(check["facts"]["运输分类"], "Port to Port") def test_inquiry_card_lists_ten_fields(self) -> None: lines = copy.inquiry_field_lines( work_order_no="WO202609200001", facts={ "起运港": "上海", "目的港": "洛杉矶", "品名": "普货", "货量": "20吨", "整柜或拼柜": "整柜", "箱型箱量": "1x40HQ", "贸易条款": "FOB", "运输分类": "Port to Port", "报价日期": "2026-09-20", }, transport_mode="海运", ) text = "\n".join(lines) self.assertIn("工单号：WO202609200001", text) self.assertIn("贸易条款：FOB", text) self.assertIn("运输分类：Port to Port", text) self.assertLess(text.index("箱型箱量"), text.index("贸易条款")) self.assertLess(text.index("贸易条款"), text.index("运输分类")) def test_clarify_asks_class_as_two_options(self) -> None: text = copy.ask_clarify( facts={ "起运港": "上海", "目的港": "洛杉矶", "品名": "服装", "货量": "1柜", "整柜或拼柜": "整柜", "箱型箱量": "1x40HQ", "贸易条款": "FOB", }, missing_keys=["运输分类"], transport_mode="海运", ) self.assertIn("1.运输分类（必填）：\n(如：Port to Port、其他）", text) self.assertIn("贸易条款：FOB", text) def test_incomplete_without_trade_terms_does_not_create_ticket(self) -> None: facts = dict(COMPLETE_SEA) facts.pop("贸易条款") phase = self.flow.on_text( sender_id="s-miss", text="海运缺条款", reply=self._reply, injected_facts=facts, injected_mode="SEA", ) self.assertEqual(phase, "clarify") self.assertIn("贸易条款", "\n".join(self.replies)) self.assertFalse((self.flow.session_of("s-miss").work_order_no or "").startswith("WO")) def test_leftover_door_to_door_not_written_as_trade_terms(self) -> None: """旧草稿里的门到门不能写到确认卡贸易条款。测服 WO202609200013。""" leftover = { "起运港": "纽约港", "目的港": "巴生西港", "品名": "机器设备", "货量": "15件", "整柜或拼柜": "整柜", "箱型箱量": "40HQ*1", "贸易条款": "门到门", } self.flow._save( FlowSession( sender_id="s-door", thread_id="s-door", phase="clarify", business_line="SEA", facts=leftover, ) ) phase = self.flow.on_text( sender_id="s-door", text="海运、纽约港到巴生西港，机器设备，40HQ*1，整柜，1.5KG，15CBM，15件", reply=self._reply, injected_facts=dict(leftover), injected_mode="SEA", ) self.assertEqual(phase, "clarify") joined = "\n".join(self.replies) self.assertNotIn("贸易条款：门到门", joined) self.assertIn("贸易条款", joined) self.assertNotEqual(self.flow.session_of("s-door").facts.get("贸易条款"), "门到门") def test_harvest_class_from_reply_then_quotes(self) -> None: first = dict(COMPLETE_SEA) first.pop("运输分类") phase = self.flow.on_text( sender_id="s-class", text="先缺分类", reply=self._reply, injected_facts=first, injected_mode="SEA", ) self.assertEqual(phase, "clarify") self.replies.clear() phase = self.flow.on_text( sender_id="s-class", text="Port to Port", reply=self._reply, injected_mode="SEA", ) self.assertEqual(phase, "wait_collab") joined = "\n".join(self.replies) self.assertIn("运输分类：Port to Port", joined) self.assertIn("贸易条款：FOB", joined) def test_complete_sea_inquiry_card_uses_ten_fields(self) -> None: phase = self.flow.on_text( sender_id="s1", text="海运完整", reply=self._reply, injected_facts=COMPLETE_SEA, injected_mode="SEA", ) self.assertEqual(phase, "wait_collab") joined = "\n".join(self.replies) self.assertIn("运输方式：海运", joined) self.assertIn("起运港：上海", joined) self.assertIn("目的港：洛杉矶", joined) self.assertIn("货物数量：20吨", joined) self.assertIn("整柜或拼柜：整柜", joined) self.assertIn("箱型箱量：1x40HQ", joined) self.assertIn("贸易条款：FOB", joined) self.assertIn("运输分类：Port to Port", joined) self.assertIn("正在查询TMS报价", joined) self.assertNotIn("件数：", joined.split("正在查询TMS报价")[0]) self.assertNotIn("本轮先开通空运", joined) sess = self.flow.session_of("s1") self.assertEqual(sess.business_line, "SEA") self.assertEqual(sess.status, "已报价") def test_new_round_after_ticket_does_not_reuse_old_facts(self) -> None: """上一单已出号，新一轮不能带上旧贸易条款。""" first = self.flow.on_text( sender_id="s-round", text="海运第一轮", reply=self._reply, injected_facts=COMPLETE_SEA, injected_mode="SEA", ) self.assertEqual(first, "wait_collab") old_no = self.flow.session_of("s-round").work_order_no self.assertTrue((old_no or "").startswith("WO")) self.replies.clear() incomplete = { "起运港": "纽约港", "目的港": "巴生西港", "品名": "机器设备", "货量": "15件", "整柜或拼柜": "整柜", "箱型箱量": "40HQ*1", "运输分类": "Port to Port", } phase = self.flow.on_text( sender_id="s-round", text="海运、纽约港到巴生西港，机器设备，40HQ*1，整柜，1.5KG，15CBM，15件，Port to Port", reply=self._reply, injected_facts=incomplete, injected_mode="SEA", ) self.assertEqual(phase, "clarify") joined = "\n".join(self.replies) self.assertNotIn("贸易条款：FOB", joined) self.assertNotIn("贸易条款：门到门", joined) self.assertIn("贸易条款", joined) sess = self.flow.session_of("s-round") self.assertNotEqual(sess.facts.get("贸易条款"), "FOB") self.assertNotEqual(sess.work_order_no, old_no) def test_handler_routes_sea_not_air_deferred(self) -> None: reset_sea_text_flow_for_test() from agent.channel.queue import MemoryMessageStore store = MemoryMessageStore() msg = InboundMessage( sender_id="s2", message_id="m-sea-1", content="海运询价", ) phase = handle_text_inquiry( msg, injected_facts=COMPLETE_SEA, injected_mode="SEA", flow=self.flow, store=store, ) self.assertEqual(phase, "wait_collab") self.assertNotEqual(phase, "sea_deferred") class SeaTmsCardTests(unittest.TestCase): def setUp(self) -> None: self.ledger = MemoryLedger() self.groups = FakeGroupClient() self.flow = SeaTextInquiryFlow(self.ledger, group_client=self.groups) self.replies: list[str] = [] self.extras: list[dict] = [] def _reply(self, text: str, extra=None) -> None: self.replies.append(text) self.extras.append(dict(extra) if extra else {}) def test_assemble_sea_query_does_not_need_air_measures(self) -> None: ready = assemble_sea_query({**COMPLETE_SEA, "贸易条款": "FOB", "运输分类": "Port to Port"}) self.assertTrue(ready["ok"]) self.assertEqual(ready["missing"], []) self.assertNotIn("件数", ready.get("missing") or []) def test_assemble_requires_trade_terms_not_class(self) -> None: src = { "起运港": "上海", "目的港": "洛杉矶", "品名": "普货", "货量": "20吨", "整柜或拼柜": "整柜", "箱型箱量": "1x40HQ", "运输分类": "Port to Port", } missing = assemble_sea_query(src) self.assertFalse(missing["ok"]) self.assertIn("贸易条款", missing["missing"]) self.assertNotIn("运输分类", missing["missing"]) src["贸易条款"] = "FOB" ready = assemble_sea_query(src) self.assertTrue(ready["ok"]) self.assertEqual(ready["tms_facts"]["贸易条款"], "FOB") self.assertNotIn("运输分类", ready["tms_facts"]) self.assertEqual(ready["facts"]["运输分类"], "Port to Port") def test_sea_quote_text_uses_three_buckets(self) -> None: quote = { "total": "6952.77", "currency": "CNY", "source_label": "TMS 标准报价", "polCostItems": [ {"name": "操作费", "amount": "1020", "currency": "CNY"}, {"name": "舱单", "amount": "50", "currency": "CNY"}, ], "oceanCostItems": [{"name": "海运费", "amount": "575", "currency": "USD"}], "podCostItems": [ {"name": "文件费", "amount": "450", "currency": "CNY"}, {"name": "设备费", "amount": "30", "currency": "CNY"}, ], "fee_total_polCostItems": "1070", "fee_total_oceanCostItems": "4038", "fee_total_podCostItems": "470", } text = copy.sea_quote_detail_text(work_order_no="WO202609200002", quote=quote) self.assertIn("始发港费用\n操作费：CNY 1020", text) self.assertIn("始发港费用小计：CNY 1070", text) self.assertIn("海运费：USD 575", text) self.assertIn("海运费小计：CNY 4038", text) self.assertIn("目的港费用小计：CNY 470", text) self.assertIn("费用合计：CNY 6952.77", text) self.assertNotIn("时效", text) def test_long_fee_lines_go_to_text_not_clipped_card(self) -> None: """海运多项费用：完整明细必须在文本里，卡片引用条不得半截。""" long_quote = { "total": "2287.65", "currency": "CNY", "eta": "1天", "valid_until": "2026-09-30", "source_label": "TMS 标准报价", "fee_lines": [ {"name": "海运费", "amount": "100.00", "currency": "USD"}, {"name": "燃油附加费", "amount": "15.00", "currency": "USD"}, {"name": "操作费", "amount": "680.00", "currency": "CNY"}, {"name": "文件费", "amount": "450.00", "currency": "CNY"}, {"name": "封条费（SEAI）", "amount": "50.00", "currency": "CNY"}, {"name": "设备费", "amount": "50.00", "currency": "CNY"}, {"name": "报关费", "amount": "250.00", "currency": "CNY"}, ], } text = copy.sea_quote_detail_text( work_order_no="WO202609150030", quote=long_quote ) self.assertEqual( text, "\n".join( [ "TMS标准报价", "WO202609150030价格明细", "海运费：USD 100.00", "燃油附加费：USD 15.00", "操作费：CNY 680.00", "文件费：CNY 450.00", "封条费（SEAI）：CNY 50.00", "设备费：CNY 50.00", "报关费：CNY 250.00", "费用合计：CNY 2287.65", ] ), ) self.assertNotIn("时效", text) self.assertNotIn("报价有效期", text) card_text = copy.quote_area_safe_text(long_quote, business_line="SEA") self.assertIn("费用合计：CNY 2287.65", card_text) self.assertNotIn("海运费：USD 100.00\n燃油附加费", card_text) self.assertLessEqual(len(card_text), 80) payload = copy.sea_tms_hit_wecom_payload( work_order_no="WO202609150030", quote=long_quote ) self.assertEqual( payload["template_card"]["quote_area"]["quote_text"], card_text ) sub = payload["template_card"]["sub_title_text"] self.assertIn("时效：1天", sub) self.assertIn("报价有效期：2026-09-30", sub) def test_sea_hit_sends_detail_text_before_card(self) -> None: """ 有价必须先发出价格明细文字，再发卡。 测服 WO202609200020：卡上写「详见上一条明细」，销售却先看到卡。 """ steps: list[str] = [] def reply(text: str, extra=None) -> str: kind = "card" if extra else "text" steps.append(kind) return f"out-{len(steps)}" def wait(out_id: str, *, work_order_no: str, what: str) -> None: steps.append(f"wait:{out_id}") self.flow._wait_outbound_sent = wait # type: ignore[method-assign] self.flow._detail_before_card_gap_sec = 0.0 self.flow._emit_sea_hit( reply, work_order_no="WO202609200020", quote={ "total": "14298.53", "currency": "CNY", "source_label": "TMS标准报价", "fee_lines": [{"name": "操作费", "amount": "1220", "currency": "CNY"}], }, ) self.assertEqual(steps[:3], ["text", "wait:out-1", "card"]) self.assertIn("card", steps) def test_sea_hit_card_has_price_and_two_buttons(self) -> None: phase = self.flow.on_text( sender_id="s1", text="海运有价", reply=self._reply, injected_facts=COMPLETE_SEA, injected_mode="SEA", ) self.assertEqual(phase, "wait_collab") joined = "\n".join(self.replies) self.assertIn("价格明细", joined) self.assertIn("是否拉产品人员进群协同", joined) self.assertIn(copy.BTN_PULL_COLLAB, joined) self.assertIn(copy.BTN_SKIP_COLLAB_TMS, joined) self.assertNotIn(copy.BTN_SKIP_COLLAB, joined) self.assertEqual(copy.canonical_button(copy.BTN_SKIP_COLLAB_TMS), copy.BTN_SKIP_COLLAB) self.assertNotIn("请手动拉相关人员", joined) self.assertNotIn("若需要咨询内部航线人员", joined) cards = [x for x in self.extras if x.get("msgtype") == "template_card"] self.assertTrue(cards) buttons = [b["text"] for b in cards[-1]["template_card"]["button_list"]] self.assertEqual(buttons, [copy.BTN_PULL_COLLAB, copy.BTN_SKIP_COLLAB_TMS]) def test_sea_miss_one_card_with_pull_button(self) -> None: facts = dict(COMPLETE_SEA) facts["目的港"] = "无价港" phase = self.flow.on_text( sender_id="s1", text="海运无价", reply=self._reply, injected_facts=facts, injected_mode="SEA", ) self.assertEqual(phase, "tms_miss") sess = self.flow.session_of("s1") self.assertEqual(sess.status, "询价中") joined = "\n".join(self.replies) self.assertIn("暂无匹配报价", joined) self.assertIn("询价中", joined) self.assertIn("未查到合适报价", joined) self.assertIn("是否需要拉内部产品人员进群", joined) self.assertIn(copy.BTN_PULL_COLLAB, joined) self.assertNotIn("请手动拉相关人员和智能体到原有群", joined) cards = [x for x in self.extras if x.get("msgtype") == "template_card"] self.assertTrue(cards) buttons = [b["text"] for b in cards[-1]["template_card"].get("button_list") or []] self.assertEqual(buttons, [copy.BTN_PULL_COLLAB]) def test_sea_tech_fail_is_not_miss(self) -> None: facts = dict(COMPLETE_SEA) facts["tms_force"] = "tech" phase = self.flow.on_text( sender_id="s1", text="海运技术失败", reply=self._reply, injected_facts=facts, injected_mode="SEA", ) self.assertEqual(phase, "system_exception") self.assertIn("TMS接口异常，IT运维排查中", "\n".join(self.replies)) self.assertNotIn(copy.BTN_PULL_COLLAB, "\n".join(self.replies)) tickets = list(self.flow._ledger._tickets.values()) self.assertTrue(tickets) self.assertEqual(tickets[-1].system_exception, "TMS异常") class SeaSkipAndPullTests(unittest.TestCase): def setUp(self) -> None: os.environ["WECOM_ARCHIVE_SEAT_USER_IDS"] = "" from agent.config import get_settings get_settings.cache_clear() self.ledger = MemoryLedger() self.ledger.set_quote_templates( [ { "templateId": "mem-default-sea", "name": "海运默认", "bizType": "海运", "keywords": [], "isDefault": True, } ] ) self.groups = FakeGroupClient() self.flow = SeaTextInquiryFlow(self.ledger, group_client=self.groups) self.replies: list[str] = [] self.extras: list[dict] = [] def _reply(self, text: str, extra=None) -> str: self.replies.append(text) self.extras.append(dict(extra) if extra else {}) return "" def _quote_first(self, sender: str = "s1") -> None: self.flow.on_text( sender_id=sender, text="海运有价", reply=self._reply, injected_facts=COMPLETE_SEA, injected_mode="SEA", ) def test_skip_collab_reuses_air_scheme(self) -> None: self._quote_first() phase = self.flow.on_button( sender_id="s1", action=copy.BTN_SKIP_COLLAB, reply=self._reply, card_meta={"response_code": "rc-sea-1"}, ) self.assertEqual(phase, "wait_adopt") ticket = self.ledger.get_ticket(work_order_no=self.flow.session_of("s1").work_order_no) self.assertEqual(ticket.event_wait_kind, "adopt") self.assertTrue(ticket.event_wait_started_at) self.assertIn("已跳过内部协同", "\n".join(self.replies)) scheme = "\n".join(x for x in self.replies if "报价方案" in x) self.assertIn("运输方式：海运", scheme) self.assertIn("货物数量：", scheme) self.assertIn("整柜或拼柜：", scheme) self.assertIn("箱型箱量：", scheme) self.assertNotIn("件数：", scheme) self.assertNotIn("重量：", scheme) self.assertNotIn("体积：", scheme) self.assertNotIn("包装类型：", scheme) cards = [ x.get("template_card") or {} for x in self.extras if (x.get("template_card") or {}).get("main_title", {}).get("title", "").endswith("报价方案") ] self.assertTrue(cards) rows = {r["keyname"]: r["value"] for r in cards[0].get("horizontal_content_list") or []} self.assertIn("海运", rows.get("航线", "")) self.assertIn("箱型", rows) self.assertNotIn("包装", rows) again = self.flow.on_button( sender_id="s1", action=copy.BTN_SKIP_COLLAB, reply=self._reply ) self.assertEqual(again, "card_discard") def test_pull_group_success_greys_buttons(self) -> None: self.ledger.set_sea_staff( [ { "name": "张产品", "wecomId": "prod1", "roleCode": "sea", "routes": "海运·全部线路", "status": "active", } ] ) self._quote_first() wo = self.flow.session_of("s1").work_order_no phase = self.flow.on_button( sender_id="s1", action=copy.BTN_PULL_COLLAB, reply=self._reply, card_meta={"response_code": "rc-pull-1"}, ) self.assertEqual(phase, "collab_group") ticket = self.ledger.get_ticket(work_order_no=wo) self.assertEqual(ticket.event_wait_kind, "product_quote") self.assertTrue(ticket.event_wait_started_at) joined = "\n".join(self.replies) self.assertIn("协同群已创建", joined) self.assertIn(wo, joined) self.assertIn("销售", joined) self.assertIn("产品张产品", joined) self.assertEqual(self.groups.created[0]["name"], wo) self.assertIn("prod1", self.groups.created[0]["userids"]) self.assertIn("s1", self.groups.created[0]["userids"]) self.assertNotIn("AI001", self.groups.created[0]["userids"]) self.assertTrue(self.groups.sent) group_text = self.groups.sent[0]["content"] self.assertTrue(any(x.get("kind") == "file" for x in self.groups.sent)) self.assertIn(wo, group_text) self.assertIn("起运港", group_text) self.assertIn("贸易条款", group_text) collab_block = group_text.split("协同补充字段：", 1)[-1].split("TMS报价：", 1)[0] self.assertNotIn("贸易条款", collab_block) self.assertIn("货好时间", collab_block) self.assertIn("商品海关编码", group_text) self.assertNotIn("HS编码", group_text) self.assertIn("TMS", group_text) sess = self.flow.session_of("s1") self.assertNotIn(copy.BTN_PULL_COLLAB, sess.allowed) self.assertNotIn(copy.BTN_SKIP_COLLAB, sess.allowed) again = self.flow.on_button( sender_id="s1", action=copy.BTN_SKIP_COLLAB, reply=self._reply ) self.assertEqual(again, "card_discard") def test_pull_prefills_spoken_no_battery(self) -> None: self.ledger.set_sea_staff( [ { "name": "张产品", "wecomId": "prod1", "roleCode": "sea", "routes": "海运·全部线路", "status": "active", } ] ) facts = dict(COMPLETE_SEA) facts["是否含电"] = "否" self.flow.on_text( sender_id="s1", text="海运不含电", reply=self._reply, injected_facts=facts, injected_mode="SEA", ) private = "\n".join(self.replies) self.assertNotIn("是否含电", private) self.flow.on_button( sender_id="s1", action=copy.BTN_PULL_COLLAB, reply=self._reply, ) group_text = "\n".join( x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text" ) self.assertIn("是否含电：否", group_text) def test_pull_sends_sheet_when_client_not_injected(self) -> None: """测服正式路径不注入 group_client，拉群仍要发出填好的模板。""" from unittest.mock import patch live = FakeGroupClient() self.ledger.set_sea_staff( [ { "name": "张产品", "wecomId": "prod1", "roleCode": "sea", "routes": "海运·全部线路", "status": "active", } ] ) with patch("agent.policy.sea_text_flow._default_group_client", return_value=live): flow = SeaTextInquiryFlow(self.ledger, group_client=None) flow.on_text( sender_id="s9", text="海运有价", reply=self._reply, injected_facts=COMPLETE_SEA, injected_mode="SEA", ) phase = flow.on_button( sender_id="s9", action=copy.BTN_PULL_COLLAB, reply=self._reply, ) self.assertEqual(phase, "collab_group") self.assertTrue(any(x.get("kind") == "text" for x in live.sent)) self.assertTrue(any(x.get("kind") == "file" for x in live.sent)) def test_pull_group_includes_archive_bot(self) -> None: self.ledger.upsert_staff( { "name": "询价机器人", "wecomId": "AI001", "roleCode": "sales", "status": "active", } ) self.ledger.set_sea_staff( [ { "name": "张产品", "wecomId": "prod1", "roleCode": "sea", "routes": "海运全线", "status": "active", } ] ) self._quote_first() self.flow.on_button( sender_id="s1", action=copy.BTN_PULL_COLLAB, reply=self._reply, ) userids = self.groups.created[0]["userids"] self.assertIn("AI001", userids) self.assertIn("s1", userids) self.assertIn("prod1", userids) joined = "\n".join(self.replies) self.assertIn("产品张产品", joined) self.assertNotIn("产品询价机器人", joined) self.assertNotIn("销售询价机器人", joined) def test_pull_no_staff_keeps_button(self) -> None: self.ledger.set_sea_staff([]) self._quote_first() phase = self.flow.on_button( sender_id="s1", action=copy.BTN_PULL_COLLAB, reply=self._reply, ) self.assertEqual(phase, "wait_collab") self.assertIn("没有可拉的产品人员", "\n".join(self.replies)) self.assertEqual(self.groups.created, []) self.assertIn(copy.BTN_PULL_COLLAB, self.flow.session_of("s1").allowed) def test_pull_already_has_group_no_second(self) -> None: self.ledger.set_sea_staff( [ { "name": "李产品", "wecomId": "prod2", "roleCode": "sea", "routes": "上海", "status": "active", } ] ) self._quote_first() self.flow.on_button( sender_id="s1", action=copy.BTN_PULL_COLLAB, reply=self._reply, ) n = len(self.groups.created) self.flow.session_of("s1").allowed = (copy.BTN_PULL_COLLAB,) phase = self.flow.on_button( sender_id="s1", action=copy.BTN_PULL_COLLAB, reply=self._reply, ) self.assertEqual(phase, "collab_group") self.assertEqual(len(self.groups.created), n) self.assertIn("群已在", "\n".join(self.replies)) def test_miss_group_message_writes_no_quote(self) -> None: self.ledger.set_sea_staff( [ { "name": "王产品", "wecomId": "prod3", "roleCode": "sea", "routes": "海运·全部线路", "status": "active", } ] ) facts = dict(COMPLETE_SEA) facts["目的港"] = "无价港" self.flow.on_text( sender_id="s1", text="海运无价", reply=self._reply, injected_facts=facts, injected_mode="SEA", ) self.flow.on_button( sender_id="s1", action=copy.BTN_PULL_COLLAB, reply=self._reply, ) self.assertTrue(self.groups.sent) group_text = self.groups.sent[0]["content"] self.assertIn("TMS 暂无匹配报价", group_text) self.assertNotIn("未查到合适报价", group_text) self.assertIn("\n\n协同补充字段：\n", group_text) self.assertIn("\n\nTMS报价：\n", group_text) self.assertNotIn("确认报价", group_text) class SeaGroupCollabFieldTests(unittest.TestCase): def setUp(self) -> None: self.ledger = MemoryLedger() self.ledger.set_quote_templates( [ { "templateId": "mem-default-sea", "name": "海运默认", "bizType": "海运", "keywords": [], "isDefault": True, } ] ) self.groups = FakeGroupClient() self.flow = SeaTextInquiryFlow(self.ledger, group_client=self.groups) self.replies: list[str] = [] def _reply(self, text: str, extra=None) -> None: self.replies.append(text) def test_group_text_ignores_trade_terms_saves_ready_date(self) -> None: self.ledger.set_sea_staff( [ { "name": "赵产品", "wecomId": "prod4", "roleCode": "sea", "routes": "海运·全部线路", "status": "active", } ] ) self.flow.on_text( sender_id="sales1", text="海运有价", reply=self._reply, injected_facts=COMPLETE_SEA, injected_mode="SEA", ) self.flow.on_button( sender_id="sales1", action=copy.BTN_PULL_COLLAB, reply=self._reply, ) wo = self.flow.session_of("sales1").work_order_no ticket = self.ledger.get_ticket(work_order_no=wo) chat_id = ticket.collab_chat_id n = len(self.replies) ignored = handle_group_collab( InboundMessage( sender_id="prod4", message_id="g1", content="贸易条款：CIF", chat_id=chat_id, chat_type="group", ), flow=self.flow, ledger=self.ledger, ) self.assertEqual(ignored, "collab_not_recognized") self.assertFalse(ticket.collab_facts.get("贸易条款")) phase = handle_group_collab( InboundMessage( sender_id="prod4", message_id="g2", content="货好时间：下周", chat_id=chat_id, chat_type="group", ), flow=self.flow, ledger=self.ledger, ) self.assertEqual(phase, "collab_field_saved") self.assertEqual(len(self.replies), n) self.assertEqual(ticket.collab_facts.get("货好时间"), "下周") def test_group_price_chat_is_ignored(self) -> None: self.ledger.set_sea_staff( [ { "name": "赵产品", "wecomId": "prod4", "roleCode": "sea", "routes": "海运·全部线路", "status": "active", } ] ) self.flow.on_text( sender_id="sales1", text="海运有价", reply=self._reply, injected_facts=COMPLETE_SEA, injected_mode="SEA", ) self.flow.on_button( sender_id="sales1", action=copy.BTN_PULL_COLLAB, reply=self._reply, ) wo = self.flow.session_of("sales1").work_order_no ticket = self.ledger.get_ticket(work_order_no=wo) phase = handle_group_collab( InboundMessage( sender_id="prod4", message_id="g2", content="报价 1200 美元可以做", chat_id=ticket.collab_chat_id, chat_type="group", ), flow=self.flow, ledger=self.ledger, ) self.assertEqual(phase, "wait_tms_choice") self.assertNotEqual(str(ticket.quote.get("total") or ""), "1200") texts = [x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text"] self.assertTrue(any("在TMS基础上改" in t and "不用TMS报价" in t for t in texts)) phase = handle_group_collab( InboundMessage( sender_id="prod4", message_id="g2b", content="按TMS改一下就行", chat_id=ticket.collab_chat_id, chat_type="group", ), flow=self.flow, ledger=self.ledger, ) self.assertEqual(phase, "product_quote_sent") ticket = self.ledger.get_ticket(work_order_no=wo) names = [str(x.get("name") or "") for x in (ticket.quote.get("fee_rows") or ticket.quote.get("fee_lines") or [])] self.assertTrue(any("空运费" in n or "海运费" in n or "附加" in n for n in names) or "1280" in str(ticket.quote.get("total") or "")) deal_texts = [x["content"] for x in self.groups.sent if x.get("kind") == "text" and "成交" in (x.get("content") or "")] self.assertTrue(deal_texts) def test_keep_tms_reads_json_string_fee_lines(self) -> None: """TMS fee_lines 常是 JSON 字符串，在TMS基础上改不能当成没明细。""" from agent.policy.sea_group_ops import merge_quote_fees base = { "source_label": "TMS 标准报价", "total": "2287.65", "currency": "CNY", "fee_lines": ( '[{"name":"海运费","amount":"100","currency":"USD","unit_price":"100"},' '{"name":"操作费","amount":"680","currency":"CNY","unit_price":"680"},' '{"name":"文件费","amount":"450","currency":"CNY","unit_price":"450"}]' ), } overlay = { "source_label": "产品报价", "total": "1200", "currency": "CNY", "fee_rows": [ {"name": "操作费", "amount": "700", "currency": "CNY"}, {"name": "文件发", "amount": "500", "currency": "CNY"}, ], } merged = merge_quote_fees(base, overlay) by_name = {str(x.get("name") or ""): x for x in (merged.get("fee_rows") or [])} self.assertEqual(str(by_name["海运费"].get("amount") or ""), "100") self.assertEqual(str(by_name["操作费"].get("amount") or ""), "700") self.assertEqual(str(by_name["操作费"].get("unit_price") or ""), "700") self.assertEqual(str(by_name["文件费"].get("amount") or ""), "500") self.assertEqual(str(by_name["文件费"].get("unit_price") or ""), "500") self.assertEqual(str(by_name["海运费"].get("unit_price") or ""), "100") self.assertNotIn("文件发", by_name) self.assertEqual(str(merged.get("total") or ""), "2357.65") self.assertNotEqual(str(merged.get("total") or ""), "1200") def test_usd_prefix_fee_lines_are_both_kept(self) -> None: """海运费：USD 100.00 / 燃油附加费：USD 15.00 两行都要抽出，不能只剩海运费。""" from agent.llm.extract_text import extract_product_quote quote = extract_product_quote("海运费：USD 100.00\n燃油附加费：USD 15.00") rows = {str(x.get("name") or ""): x for x in (quote.get("fee_rows") or [])} self.assertIn("海运费", rows) self.assertIn("燃油附加费", rows) self.assertEqual(str(rows["海运费"].get("amount") or ""), "100.00") self.assertEqual(str(rows["燃油附加费"].get("amount") or ""), "15.00") def test_air_unit_qty_amount_keeps_air_freight_name(self) -> None: """空运费 单价/数量/金额：费用名是空运费，金额用「金额」不是单价。""" from agent.llm.extract_text import extract_product_quote quote = extract_product_quote( "空运段，空运费 单价CNY99.00，数量100 ，金额CNY199.00@询价小助手" ) rows = {str(x.get("name") or ""): x for x in (quote.get("fee_rows") or [])} self.assertIn("空运费", rows) self.assertNotIn("海运费", rows) self.assertEqual(str(rows["空运费"].get("amount") or ""), "199.00") self.assertEqual(str(rows["空运费"].get("currency") or ""), "CNY") self.assertEqual(str(rows["空运费"].get("unit_price") or ""), "99.00") self.assertEqual(str(quote.get("total") or ""), "199.0") def test_fee_line_text_is_a_product_quote(self) -> None: """产品按行改价：码头操作费100CNY / 报关费改为300CNY，必须抽出费用名，不能只当海运费合计。""" from agent.llm.extract_text import extract_product_quote quote = extract_product_quote("码头操作费100CNY\n报关费改为300CNY") rows = {str(x.get("name") or ""): x for x in (quote.get("fee_rows") or [])} self.assertIn("码头操作费", rows) self.assertIn("报关费", rows) self.assertEqual(str(rows["码头操作费"].get("amount") or ""), "100") self.assertEqual(str(rows["报关费"].get("amount") or ""), "300") self.assertNotEqual( [str(x.get("name") or "") for x in (quote.get("fee_rows") or [])], ["海运费"], ) def test_fee_change_typo_and_colon(self) -> None: """操作费改为700 + 文件发改为：500，两行都要抽出，文件发要对上文件费。""" from agent.llm.extract_text import extract_product_quote from agent.policy.sea_group_ops import _fee_name_same, merge_quote_fees quote = extract_product_quote("操作费改为700CNY\n文件发改为：500CNY") rows = {str(x.get("name") or ""): x for x in (quote.get("fee_rows") or [])} self.assertEqual(str(rows["操作费"].get("amount") or ""), "700") self.assertEqual(str(rows["文件发"].get("amount") or ""), "500") self.assertTrue(_fee_name_same("文件发", "文件费")) self.assertFalse(_fee_name_same("操作费", "码头操作费")) merged = merge_quote_fees( { "source_label": "TMS标准报价", "fee_rows": [ {"name": "操作费", "amount": "680", "currency": "CNY"}, {"name": "文件费", "amount": "450", "currency": "CNY"}, ], }, quote, ) by_name = {str(x.get("name") or ""): x for x in (merged.get("fee_rows") or [])} self.assertEqual(str(by_name["操作费"].get("amount") or ""), "700") self.assertEqual(str(by_name["文件费"].get("amount") or ""), "500") self.assertNotIn("文件发", by_name) class SeaContinueTests(unittest.TestCase): def setUp(self) -> None: from agent.redis_coord.runtime import shutdown_redis_runtime shutdown_redis_runtime() self.ledger = MemoryLedger() self.ledger.set_quote_templates( [ { "templateId": "mem-default-sea", "name": "海运默认", "bizType": "海运", "keywords": [], "isDefault": True, } ] ) self.flow = SeaTextInquiryFlow(self.ledger) self.replies: list[str] = [] self.extras: list[dict] = [] def _reply(self, text: str, extra=None) -> str: self.replies.append(text) self.extras.append(dict(extra) if extra else {}) return "" def test_continue_after_skip_replays_sea_scheme_not_air(self) -> None: """已跳过协同后再续办：必须海运方案卡，不能套空运件数包装。""" from unittest.mock import patch from agent.channel.queue import MemoryMessageStore from agent.handlers.continue_thread import handle_continue_thread from agent.policy.air_text_flow import AirTextInquiryFlow self.flow.on_text( sender_id="c1", text="海运有价", reply=self._reply, injected_facts=COMPLETE_SEA, injected_mode="SEA", ) self.flow.on_button( sender_id="c1", action=copy.BTN_SKIP_COLLAB, reply=self._reply, card_meta={"response_code": "rc-cont-1"}, ) wo = self.flow.session_of("c1").work_order_no store = MemoryMessageStore() msg = InboundMessage( sender_id="c1", message_id="m-cont-sea", content=f"继续工单{wo}", ) air_trap = AirTextInquiryFlow(MemoryLedger()) with patch( "agent.handlers.continue_thread.get_sea_text_flow", return_value=self.flow ), patch( "agent.handlers.continue_thread.get_air_text_flow", return_value=air_trap ): phase = handle_continue_thread(msg, thread_id=wo, store=store) self.assertEqual(phase, "wait_adopt") cards = [ (item.payload or {}).get("template_card") or {} for item in store._outbox.values() ] titles = [ (c.get("main_title") or {}).get("title") or "" for c in cards if c ] self.assertTrue(any(t == f"{wo}报价方案" for t in titles)) rows = {} for card in cards: if (card.get("main_title") or {}).get("title") == f"{wo}报价方案": rows = { r["keyname"]: r["value"] for r in card.get("horizontal_content_list") or [] } self.assertIn("海运", rows.get("航线", "")) self.assertIn("箱型", rows) self.assertNotIn("包装", rows) def test_continue_from_ledger_uses_sea_collab_card(self) -> None: """进程无书签：主账海运已报价应回收海运协同卡，不是空运卡。""" from unittest.mock import patch from agent.channel.queue import MemoryMessageStore from agent.handlers.continue_thread import handle_continue_thread from agent.policy.air_text_flow import AirTextInquiryFlow created = self.ledger.create_ticket( sender_id="c2", business_line="SEA", facts=COMPLETE_SEA ) wo = created["work_order_no"] self.ledger.upsert_quote( work_order_no=wo, quote={ "total": "USD 1,200", "eta": "18天", "valid_until": "2026-09-20", "source_label": "TMS 标准报价", }, to_status="已报价", ) empty = SeaTextInquiryFlow(self.ledger) store = MemoryMessageStore() msg = InboundMessage( sender_id="c2", message_id="m-cont-ledger", content=f"继续工单{wo}", ) air_trap = AirTextInquiryFlow(MemoryLedger()) with patch( "agent.handlers.continue_thread.get_sea_text_flow", return_value=empty ), patch( "agent.handlers.continue_thread.get_air_text_flow", return_value=air_trap ): phase = handle_continue_thread(msg, thread_id=wo, store=store) self.assertEqual(phase, "wait_collab") sess = empty.session_by_ticket("c2", wo) self.assertIsNotNone(sess) self.assertEqual(sess.business_line, "SEA") joined = "\n".join(item.content for item in store._outbox.values()) self.assertIn(copy.BTN_PULL_COLLAB, joined) self.assertIn("是否拉产品人员进群协同", joined) self.assertNotIn("包装方式", joined) def test_continue_after_quote_file_returns_deal(self) -> None: """报价单已发出后再续办：必须成交跟进，不能退回生成报价单。""" from unittest.mock import patch from agent.channel.queue import MemoryMessageStore from agent.handlers.continue_thread import handle_continue_thread from agent.policy.air_text_flow import AirTextInquiryFlow self.flow.on_text( sender_id="c3", text="海运有价", reply=self._reply, injected_facts=COMPLETE_SEA, injected_mode="SEA", ) self.flow.on_button( sender_id="c3", action=copy.BTN_SKIP_COLLAB, reply=self._reply, card_meta={"response_code": "rc-deal-1"}, ) wo = self.flow.session_of("c3").work_order_no self.flow.on_button( sender_id="c3", action=copy.BTN_ADOPT_EXCEL, reply=self._reply, card_meta={"event_key": f"adopt_excel:{wo}", "response_code": "rc-xl"}, ) self.assertEqual(self.flow.session_of("c3").phase, "wait_deal") file_extras = [x for x in self.extras if x.get("msgtype") == "file"] self.assertTrue(file_extras) follow = file_extras[-1].get("followup_card") or {} self.assertTrue(str(follow.get("task_id") or "").startswith("deal-")) separate_deal = [ x for x in self.extras if x.get("msgtype") == "template_card" and str((x.get("template_card") or {}).get("task_id") or "").startswith("deal-") ] self.assertEqual(separate_deal, []) empty = SeaTextInquiryFlow(self.ledger) store = MemoryMessageStore() msg = InboundMessage( sender_id="c3", message_id="m-cont-deal", content=f"继续工单{wo}", ) air_trap = AirTextInquiryFlow(MemoryLedger()) with patch( "agent.handlers.continue_thread.get_sea_text_flow", return_value=empty ), patch( "agent.handlers.continue_thread.get_air_text_flow", return_value=air_trap ): phase = handle_continue_thread(msg, thread_id=wo, store=store) self.assertEqual(phase, "wait_deal") joined = "\n".join(item.content for item in store._outbox.values()) self.assertFalse(any(x in joined for x in (copy.BTN_DEAL, "成交"))) self.assertNotIn(copy.BTN_SKIP_COLLAB, joined) class SeaGroupCopyTests(unittest.TestCase): def test_group_brief_uses_customs_code_and_section_breaks(self) -> None: text = copy.sea_group_brief( work_order_no="WO202609150023", facts={ "起运港": "纽约港", "目的港": "巴生西港", "品名": "机器设备", "货量": "15", "整柜或拼柜": "整柜", "箱型箱量": "40HQ*1", "报价日期": "2026-09-15", }, quote=None, collab_facts={}, sales_name="吴季林", product_names=["罗小华"], has_price=False, ) self.assertIn("@吴季林 @罗小华", text) self.assertIn("询价字段：\n运输方式：海运", text) self.assertIn("\n\n协同补充字段：\n", text) self.assertIn("商品海关编码：-", text) self.assertIn("货好时间：-", text) collab_block = text.split("协同补充字段：", 1)[-1].split("TMS报价：", 1)[0] self.assertNotIn("贸易条款", collab_block) self.assertNotIn("HS编码", text) self.assertIn("\n\nTMS报价：\nTMS 暂无匹配报价", text) self.assertIn("产品人员可通过“纯文字”告知新报价，或直接将新报价excel表发给智能体。", text) self.assertIn("附件报价单仅作为文件存储于后台，智能体不读取且不将报价内容写进系统", text) self.assertNotIn("航线人员可通过", text) self.assertNotIn("未查到合适报价", text) self.assertNotIn("确认报价", text) def test_group_brief_has_confirm_hint_when_tms_priced(self) -> None: text = copy.sea_group_brief( work_order_no="WO202609160004", facts={"起运港": "DONGGUAN", "目的港": "海防", "品名": "机器设备"}, quote={"total": "2287.65", "currency": "CNY", "valid_until": "2026-09-30"}, collab_facts={}, sales_name="吴季林", product_names=["林绵勋"], has_price=True, ) self.assertIn("报价有效期：2026-09-30", text) self.assertIn("请产品人员回复“确认报价”确定采用TMS报价。", text) self.assertIn("亦可通过“纯文字”告知新报价，或直接将新报价excel表发给智能体。", text) self.assertIn("附件报价单仅作为文件存储于后台，智能体不读取且不将报价内容写进系统", text) self.assertNotIn("请航线人员回复", text) self.assertNotIn("人工报价：", text) def test_group_brief_manual_quote_stays_off_tms(self) -> None: text = copy.sea_group_brief( work_order_no="WO202609160006", facts={"起运港": "上海港", "目的港": "海防", "品名": "机器设备"}, quote={ "source_label": "产品报价", "total": "115", "currency": "USD", "fee_rows": [ {"name": "海运费", "amount": "100.00", "currency": "USD"}, {"name": "燃油附加费", "amount": "15.00", "currency": "USD"}, ], }, collab_facts={}, sales_name="吴季林", product_names=["林绵勋", "罗小华", "吴季林"], has_price=True, tms_quote={}, product_quote={ "source_label": "产品报价", "fee_rows": [ {"name": "海运费", "amount": "100.00", "currency": "USD"}, {"name": "燃油附加费", "amount": "15.00", "currency": "USD"}, ], }, ) self.assertIn("TMS报价：\nTMS 暂无匹配报价", text) self.assertIn("产品人员可通过“纯文字”告知新报价，或直接将新报价excel表发给智能体。", text) self.assertIn("人工报价：\n海运费：USD 100.00\n燃油附加费：USD 15.00", text) self.assertNotIn("确认报价", text) self.assertNotIn("总销售价合计：USD 100.00", text) def test_scheme_card_uses_sea_fields(self) -> None: facts = { "起运港": "上海", "目的港": "洛杉矶", "品名": "普货", "货量": "15件", "整柜或拼柜": "整柜", "箱型箱量": "1x40HQ", "报价日期": "2026-09-15", } quote = {"total": "USD 1200", "eta": "18天", "valid_until": "2026-09-20"} text = copy.scheme_card( work_order_no="WO202609150021", status="已报价", facts=facts, quote=quote, transport_mode="海运", ) self.assertIn("运输方式：海运", text) self.assertIn("货物数量：15件", text) self.assertIn("箱型箱量：1x40HQ", text) self.assertNotIn("件数：", text) self.assertNotIn("包装类型：", text) rows = { x["keyname"]: x["value"] for x in copy.scheme_horizontal_rows( work_order_no="WO202609150021", facts=facts, quote=quote, transport_mode="海运", ) } self.assertIn("海运 上海→洛杉矶", rows["航线"]) self.assertIn("15件", rows["货物"]) self.assertIn("1x40HQ", rows["箱型"]) self.assertNotIn("包装", rows) def test_lost_ack_asks_reason_only(self) -> None: text = copy.deal_ack("未成交", "WO202609160004") self.assertEqual(text, "已记录为「未成交」。请补充未成交原因。") self.assertNotIn("将关闭智能体跟进", text) def test_invited_names_carry_role(self) -> None: text = copy.sea_group_created( work_order_no="WO202609150019", sales_name="吴季林", product_names=["罗小华"], ) self.assertEqual( text, "协同群已创建，群名与工单号一致：WO202609150019。" "已邀请：销售吴季林、产品罗小华。请在群聊中继续确认报价。", ) def test_group_only_excel_pdf_copy(self) -> None: self.assertEqual( copy.group_only_excel_pdf(), "群里只收 Excel 和 PDF，其它文件不会存入工单", ) class SeaGroupOpsTests(unittest.TestCase): """群协同：附件类型、产品报价、解散后再拉、终态提醒。""" def setUp(self) -> None: self.ledger = MemoryLedger() self.ledger.set_quote_templates( [ { "templateId": "mem-default-sea", "name": "海运默认", "bizType": "海运", "keywords": [], "isDefault": True, } ] ) self.ledger.set_sea_staff( [ { "name": "赵产品", "wecomId": "prod4", "roleCode": "sea", "routes": "海运·全部线路", "status": "active", } ] ) self.ledger.upsert_staff({"name": "吴吉林", "wecomId": "sales1"}) self.groups = FakeGroupClient() self.flow = SeaTextInquiryFlow(self.ledger, group_client=self.groups) self.replies: list[str] = [] def _reply(self, text: str, extra=None) -> None: self.replies.append(text) def _open_group(self) -> tuple[str, str]: self.flow.on_text( sender_id="sales1", text="海运有价", reply=self._reply, injected_facts=COMPLETE_SEA, injected_mode="SEA", ) self.flow.on_button( sender_id="sales1", action=copy.BTN_PULL_COLLAB, reply=self._reply, ) sess = self.flow.session_of("sales1") return sess.work_order_no, sess.collab_chat_id def test_group_rejects_non_excel_pdf(self) -> None: wo, chat = self._open_group() phase = handle_group_collab( InboundMessage( sender_id="sales1", message_id="img1", content="", msg_type="image", chat_id=chat, chat_type="group", media={"filename": "现场.jpg", "media_id": "mid-1"}, ), flow=self.flow, ledger=self.ledger, ) self.assertEqual(phase, "file_type_rejected") texts = [x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text"] self.assertIn("群里只收 Excel 和 PDF，其它文件不会存入工单", texts) ticket = self.ledger.get_ticket(work_order_no=wo) self.assertFalse(ticket.attachments) def test_sales_excel_is_archived_not_quote(self) -> None: wo, chat = self._open_group() phase = handle_group_collab( InboundMessage( sender_id="sales1", message_id="xls1", content="", msg_type="file", chat_id=chat, chat_type="group", media={"filename": "对账单.xlsx", "media_id": "mid-2"}, ), flow=self.flow, ledger=self.ledger, ) self.assertEqual(phase, "file_archived") ticket = self.ledger.get_ticket(work_order_no=wo) self.assertEqual(ticket.attachments[0]["file_name"], "对账单.xlsx") self.assertNotEqual(ticket.quote.get("source_label"), "产品附件报价") def test_product_excel_quotes_and_deal(self) -> None: wo, chat = self._open_group() phase = handle_group_collab( InboundMessage( sender_id="prod4", message_id="xls2", content="", msg_type="file", chat_id=chat, chat_type="group", media={"filename": "产品报价.pdf", "media_id": "mid-3"}, ), flow=self.flow, ledger=self.ledger, ) self.assertEqual(phase, "product_file_quote") ticket = self.ledger.get_ticket(work_order_no=wo) self.assertEqual(ticket.event_wait_kind, "deal") self.assertTrue(ticket.event_wait_started_at) self.assertEqual(ticket.attachments[0]["file_name"], "产品报价.pdf") self.assertEqual(ticket.status, "已报价") texts = [x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text"] self.assertIn(f"工单{wo}的1个附件已保存到后台工单", texts) self.assertNotIn("附件已保存后台", texts) ask = [t for t in texts if "在TMS基础上改" in t] self.assertFalse(ask) self.assertTrue(any(x.get("kind") == "card" for x in self.groups.sent)) def test_terminal_file_speaks(self) -> None: wo, chat = self._open_group() self.ledger.transition(work_order_no=wo, to_status="已成交") sess = self.flow.session_of("sales1") sess.status = "已成交" phase = handle_group_collab( InboundMessage( sender_id="prod4", message_id="xls3", content="", msg_type="file", chat_id=chat, chat_type="group", media={"filename": "晚到.xlsx", "media_id": "mid-4"}, ), flow=self.flow, ledger=self.ledger, ) self.assertEqual(phase, "ticket_locked_file") texts = [x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text"] self.assertTrue(any("不能再上传附件" in t for t in texts)) self.assertFalse(self.ledger.get_ticket(work_order_no=wo).attachments) def test_handoff_group_stays_silent(self) -> None: from agent.routing.group_intent import INTENT_FIELDS wo, chat = self._open_group() self.ledger.transition(work_order_no=wo, to_status="转人工") sess = self.flow.session_of("sales1") sess.status = "转人工" sess.handoff_committed = True self.flow._save(sess) before = len(self.groups.sent) phase = handle_group_collab( InboundMessage( sender_id="prod4", message_id="h1", content="客户名称张三", chat_id=chat, chat_type="group", ), flow=self.flow, ledger=self.ledger, injected_intent=INTENT_FIELDS, ) self.assertEqual(phase, "handoff_ignore") after = [x.get("content") or "" for x in self.groups.sent[before:] if x.get("kind") == "text"] self.assertFalse(any("不能修改工单信息" in t for t in after)) ticket = self.ledger.get_ticket(work_order_no=wo) self.assertNotIn("客户名称", ticket.collab_facts or {}) def test_dissolve_then_cannot_recreate(self) -> None: wo, chat = self._open_group() phase = handle_group_collab( InboundMessage( sender_id="sales1", message_id="dis1", content="这个群解散吧", chat_id=chat, chat_type="group", ), flow=self.flow, ledger=self.ledger, injected_intent="dissolve", ) self.assertEqual(phase, "group_dissolved") self.assertTrue(any(x.get("kind") == "dismiss" for x in self.groups.sent)) self.assertTrue(self.ledger.get_ticket(work_order_no=wo).collab_ended) sess = self.flow.session_of("sales1") sess.allowed = (copy.BTN_PULL_COLLAB,) again = self.flow.on_button( sender_id="sales1", action=copy.BTN_PULL_COLLAB, reply=self._reply, ) self.assertEqual(again, "collab_ended") self.assertIn("不能再拉群", "\n".join(self.replies)) def test_group_oral_keeps_inquiry_facts_when_snapshot_skinny(self) -> None: wo, chat = self._open_group() real_find = self.ledger.find_by_collab_chat def skinny(*, chat_id: str): ticket = real_find(chat_id=chat_id) if ticket is None: return None return { "found": True, "work_order_no": ticket.work_order_no, "sales_wecom_id": ticket.sales_wecom_id, "collab_chat_id": ticket.collab_chat_id, "collab_member_ids": list(ticket.collab_member_ids), "collab_facts": dict(ticket.collab_facts), } self.ledger.find_by_collab_chat = skinny # type: ignore[method-assign] self.flow._by_ticket.clear() self.flow._by_sender.clear() phase = handle_group_collab( InboundMessage( sender_id="sales1", message_id="oral-keep-facts-1", content="不含电、不含油、随时可提", chat_id=chat, chat_type="group", ), flow=self.flow, ledger=self.ledger, injected_intent="collab_fields", ) self.assertEqual(phase, "collab_field_saved") texts = "\n".join(x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text") self.assertIn("上海", texts) self.assertIn("洛杉矶", texts) self.assertIn("随时可提", texts) self.assertNotRegex(texts, r"起运港：\s*-") self.assertIn("@吴吉林 @赵产品", texts) self.assertNotRegex(texts, r"^@销售(\s|$)") briefs = [ x for x in self.groups.sent if x.get("kind") == "text" and "询价字段" in (x.get("content") or "") ] self.assertTrue(briefs) self.assertIn("sales1", briefs[-1].get("mention_userids") or []) self.assertIn("prod4", briefs[-1].get("mention_userids") or []) self.assertNotIn("AI001", briefs[-1].get("mention_userids") or []) def test_group_oral_oil_power_ready_is_saved(self) -> None: wo, chat = self._open_group() n = len([x for x in self.groups.sent if x.get("kind") == "text"]) phase = handle_group_collab( InboundMessage( sender_id="sales1", message_id="oral-collab-1", content="不含电、不含油、随时可提", chat_id=chat, chat_type="group", ), flow=self.flow, ledger=self.ledger, injected_intent="collab_fields", ) self.assertEqual(phase, "collab_field_saved") ticket = self.ledger.get_ticket(work_order_no=wo) self.assertEqual(ticket.collab_facts.get("是否含电"), "否") self.assertEqual(ticket.collab_facts.get("是否含油"), "否") self.assertEqual(ticket.collab_facts.get("货好时间"), "随时可提") texts = [x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text"] self.assertGreater(len(texts), n) def test_collab_facts_are_projected_for_quote_sheet(self) -> None: from agent.policy.quote_overlay import overlay_labeled_facts from agent.policy.sea_group_ops import _facts_for_fill from openpyxl import Workbook, load_workbook filled = _facts_for_fill( {"起运港": "DONGGUAN", "目的港": "海防"}, {"货好时间": "随时可提", "是否含油": "否", "是否含电": "否", "HS编码": "8479"}, ) self.assertEqual(filled["cargo_ready_date"], "随时可提") self.assertEqual(filled["hs_code"], "8479") self.assertIn("油：否", filled["oil_elec_mag"]) self.assertIn("电：否", filled["oil_elec_mag"]) import tempfile from pathlib import Path wb = Workbook() ws = wb.active ws["A1"] = "*货好时间/Cargo Ready Date：" ws["A2"] = "*商品海关编码/HS Code：" ws["A3"] = "*是否含油,电,磁/OIL,ELEC,MAG：" path = str(Path(tempfile.mkdtemp(prefix="collab-sheet-")) / "src.xlsx") wb.save(path) out = overlay_labeled_facts(filepath=path, facts=filled) painted = load_workbook(out["filepath"]) sheet = painted.active self.assertEqual(sheet["B1"].value, "随时可提") self.assertEqual(sheet["B2"].value, "8479") self.assertIn("油：否", str(sheet["B3"].value or "")) def test_group_deal_text_asks_to_type(self) -> None: """群里成交跟进是文字版：@销售、请确认、请输入，不要报价合计和请选择。""" text = copy.group_deal_followup(sales_name="吴季林", work_order_no="WO202609160001") self.assertEqual( text, ( "@吴季林\n" "询价工单 WO202609160001\n" "报价单已发送，请确认成交状态\n" "咨询成交状态：请输入成交 / 未成交 (未成交需补充原因)/ 协商中" ), ) wo, chat = self._open_group() handle_group_collab( InboundMessage( sender_id="prod4", message_id="xls-deal-copy", content="", msg_type="file", chat_id=chat, chat_type="group", media={"filename": "产品报价.pdf", "media_id": "mid-deal-copy"}, ), flow=self.flow, ledger=self.ledger, ) texts = [x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text"] follow = [t for t in texts if "请确认成交状态" in t and "请输入成交" in t] self.assertTrue(follow) self.assertTrue(any(t.startswith("@") and f"询价工单 {wo}" in t for t in follow)) self.assertFalse(any("请选择成交" in t or "请跟进成交状态" in t or "总销售价合计" in t for t in follow)) def test_group_text_deal_does_not_go_to_private_card(self) -> None: """群里打「成交」不能当私聊点卡丢掉，必须走群成交。""" from unittest.mock import patch from agent.routing.decision import RouteDecision from agent.routing.dispatch import dispatch_inbound wo, chat = self._open_group() with patch("agent.handlers.group_collab.get_sea_text_flow", return_value=self.flow): result = dispatch_inbound( InboundMessage( sender_id="sales1", message_id="deal-dispatch-1", content="成交", chat_id=chat, chat_type="group", ), RouteDecision(intent="ordinary_text_other"), ) self.assertEqual(result, "group_collab") ticket = self.ledger.get_ticket(work_order_no=wo) self.assertEqual(ticket.status, "已成交") def test_sales_can_close_deal_after_skinny_snapshot(self) -> None: """按群找回的工单只有字段、没有报价时，未成交也要能入账。""" wo, chat = self._open_group() real_find = self.ledger.find_by_collab_chat def skinny(*, chat_id: str): ticket = real_find(chat_id=chat_id) if ticket is None: return None return { "found": True, "work_order_no": ticket.work_order_no, "sales_wecom_id": ticket.sales_wecom_id, "collab_chat_id": ticket.collab_chat_id, "collab_member_ids": list(ticket.collab_member_ids), "facts": dict(ticket.facts), } self.ledger.find_by_collab_chat = skinny # type: ignore[method-assign] self.flow._by_ticket.clear() self.flow._by_sender.clear() phase = handle_group_collab( InboundMessage( sender_id="sales1", message_id="lost-skinny-1", content="未成交", chat_id=chat, chat_type="group", ), flow=self.flow, ledger=self.ledger, ) self.assertEqual(phase, "wait_lost_reason") self.assertEqual(self.ledger.get_ticket(work_order_no=wo).status, "未成交") def test_sales_next_text_is_lost_reason(self) -> None: wo, chat = self._open_group() handle_group_collab( InboundMessage( sender_id="sales1", message_id="lost-1", content="未成交", chat_id=chat, chat_type="group", ), flow=self.flow, ledger=self.ledger, ) phase = handle_group_collab( InboundMessage( sender_id="sales1", message_id="lost-2", content="客人比价了", chat_id=chat, chat_type="group", ), flow=self.flow, ledger=self.ledger, ) self.assertEqual(phase, "done") self.assertEqual(self.ledger.get_ticket(work_order_no=wo).lost_reason, "客人比价了") def test_sales_lost_and_reason_same_message(self) -> None: """群里一句同时说未成交和原因，不再追问。""" wo, chat = self._open_group() phase = handle_group_collab( InboundMessage( sender_id="sales1", message_id="lost-inline-1", content="未成交，原因客户不想要了", chat_id=chat, chat_type="group", ), flow=self.flow, ledger=self.ledger, ) self.assertEqual(phase, "done") ticket = self.ledger.get_ticket(work_order_no=wo) self.assertEqual(ticket.status, "未成交") self.assertEqual(ticket.lost_reason, "客户不想要了") def test_sales_can_close_deal_by_text(self) -> None: """群里销售说成交，要入账并回话，不能装没听见。相近说法也认。""" wo, chat = self._open_group() phase = handle_group_collab( InboundMessage( sender_id="sales1", message_id="deal-text-1", content="成交", chat_id=chat, chat_type="group", ), flow=self.flow, ledger=self.ledger, ) self.assertEqual(phase, "done") ticket = self.ledger.get_ticket(work_order_no=wo) self.assertEqual(ticket.status, "已成交") texts = [x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text"] self.assertTrue(any("已记录为「已成交」" in t for t in texts)) def test_product_cannot_close_deal_by_text(self) -> None: wo, chat = self._open_group() phase = handle_group_collab( InboundMessage( sender_id="prod4", message_id="deal-text-2", content="这个成了", chat_id=chat, chat_type="group", ), flow=self.flow, ledger=self.ledger, ) self.assertEqual(phase, "not_sales") ticket = self.ledger.get_ticket(work_order_no=wo) self.assertNotEqual(ticket.status, "已成交") texts = [x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text"] self.assertTrue(any("只有负责该工单的销售" in t for t in texts)) def test_stale_deal_card_speaks(self) -> None: from agent.policy.sea_group_ops import explain_deal_click self._open_group() sess = self.flow.session_of("sales1") sess.deal_version = 2 sess.collab_chat_id = sess.collab_chat_id or "wr_sea_1" reason = explain_deal_click( flow=self.flow, sender_id="sales1", action=copy.BTN_DEAL, card_meta={"event_key": f"deal:{sess.work_order_no}:v1"}, sess=sess, ) self.assertEqual(reason, "stale_deal") texts = [x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text"] self.assertTrue(any("已过期" in t for t in texts)) def test_sales_who_is_also_product_can_quote(self) -> None: """拉群销售若后台也是产品·海运，群里按行改价必须收下，不能装没看见。""" self.ledger.set_sea_staff( [ { "name": "吴吉林", "wecomId": "sales1", "roleCode": "sea", "routes": "海运·全部线路", "status": "active", }, { "name": "赵产品", "wecomId": "prod4", "roleCode": "sea", "routes": "海运·全部线路", "status": "active", }, ] ) wo, chat = self._open_group() ticket = self.ledger.get_ticket(work_order_no=wo) self.assertIn("sales1", ticket.product_ids) phase = handle_group_collab( InboundMessage( sender_id="sales1", message_id="dual-role-quote", content="码头操作费100CNY\n报关费改为300CNY", chat_id=chat, chat_type="group", ), flow=self.flow, ledger=self.ledger, injected_intent="product_quote", ) self.assertEqual(phase, "wait_tms_choice") def test_existing_ticket_sales_later_product_role_can_quote(self) -> None: """已建群后销售才挂上产品岗：仍按后台岗位认，不能只看建群时的产品名单。""" wo, chat = self._open_group() ticket = self.ledger.get_ticket(work_order_no=wo) self.assertNotIn("sales1", ticket.product_ids) self.ledger.set_sea_staff( [ { "name": "吴吉林", "wecomId": "sales1", "roleCode": "sea", "routes": "海运·全部线路", "status": "active", } ] ) phase = handle_group_collab( InboundMessage( sender_id="sales1", message_id="later-product-quote", content="码头操作费100CNY\n报关费改为300CNY", chat_id=chat, chat_type="group", ), flow=self.flow, ledger=self.ledger, injected_intent="product_quote", ) self.assertEqual(phase, "wait_tms_choice") def test_product_fee_line_quote_asks_then_keeps_tms(self) -> None: """有 TMS 价时先问；选「在基础上改」才出单，没点名的 TMS 费用保留。""" wo, chat = self._open_group() ticket = self.ledger.get_ticket(work_order_no=wo) ticket.quote = { "total": "2287.65", "currency": "CNY", "source_label": "TMS 标准报价", "fee_lines": [ {"name": "海运费", "amount": "100.00", "currency": "USD"}, {"name": "操作费", "amount": "680.00", "currency": "CNY"}, {"name": "报关费", "amount": "250.00", "currency": "CNY"}, ], } phase = handle_group_collab( InboundMessage( sender_id="prod4", message_id="fee-line-1", content="码头操作费100CNY\n报关费改为300CNY", chat_id=chat, chat_type="group", ), flow=self.flow, ledger=self.ledger, injected_intent="product_quote", ) self.assertEqual(phase, "wait_tms_choice") texts = [x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text"] self.assertTrue(any("在TMS基础上改" in t and "不用TMS报价" in t for t in texts)) self.assertFalse(any("没有识别到报价" in t for t in texts)) ticket = self.ledger.get_ticket(work_order_no=wo) before = {str(x.get("name") or ""): x for x in (ticket.quote.get("fee_lines") or [])} self.assertEqual(str(before["报关费"].get("amount") or ""), "250.00") phase = handle_group_collab( InboundMessage( sender_id="prod4", message_id="fee-line-2", content="在TMS基础上改", chat_id=chat, chat_type="group", ), flow=self.flow, ledger=self.ledger, ) self.assertEqual(phase, "product_quote_sent") ticket = self.ledger.get_ticket(work_order_no=wo) names = { str(x.get("name") or ""): x for x in (ticket.quote.get("fee_lines") or ticket.quote.get("fee_rows") or []) } self.assertEqual(str(names["报关费"].get("amount") or ""), "300") self.assertEqual(str(names["操作费"].get("amount") or ""), "680.00") self.assertEqual(str(names["码头操作费"].get("amount") or ""), "100") self.assertEqual(str(names["海运费"].get("amount") or ""), "100.00") def test_product_text_on_miss_is_manual_not_tms(self) -> None: """TMS 无价时产品文字报价：摘要 TMS 仍是暂无，人工报价两行齐全。""" facts = dict(COMPLETE_SEA) facts["目的港"] = "无价港" self.flow.on_text( sender_id="sales1", text="海运无价", reply=self._reply, injected_facts=facts, injected_mode="SEA", ) self.flow.on_button( sender_id="sales1", action=copy.BTN_PULL_COLLAB, reply=self._reply, ) sess = self.flow.session_of("sales1") chat = sess.collab_chat_id phase = handle_group_collab( InboundMessage( sender_id="prod4", message_id="miss-manual-1", content="海运费：USD 100.00\n燃油附加费：USD 15.00", chat_id=chat, chat_type="group", ), flow=self.flow, ledger=self.ledger, ) self.assertEqual(phase, "product_quote_sent") briefs = [x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text"] text = "\n".join(briefs) self.assertIn("TMS 暂无匹配报价", text) self.assertIn("人工报价：", text) self.assertIn("海运费：USD 100.00", text) self.assertIn("燃油附加费：USD 15.00", text) self.assertNotIn("确认报价", text) def test_confirm_tms_saves_sheet_and_notifies_app(self) -> None: """有 TMS 价时产品回确认报价：模板进后台，群里告诉大家已保存。""" from agent.routing.group_intent import INTENT_CONFIRM_TMS, classify_group_text self.assertEqual(classify_group_text("确认报价", allow_network=False), INTENT_CONFIRM_TMS) wo, chat = self._open_group() phase = handle_group_collab( InboundMessage( sender_id="prod4", message_id="confirm-tms-1", content="确认报价", chat_id=chat, chat_type="group", ), flow=self.flow, ledger=self.ledger, ) self.assertEqual(phase, "quote_confirmed") ticket = self.ledger.get_ticket(work_order_no=wo) self.assertEqual(ticket.status, "已报价") self.assertTrue(ticket.attachments) self.assertIn("quote", str(ticket.attachments[0].get("file_name") or "").lower()) texts = [x.get("content") or "" for x in self.groups.sent if x.get("kind") == "text"] self.assertIn(f"工单{wo}的1个附件已保存到后台工单", texts) self.assertTrue(any("成交" in t for t in texts)) def test_first_option_text_keeps_tms(self) -> None: """问句列了 1/2 时，回「第一个」要按在 TMS 基础上改，不能装没听见。""" from agent.routing.group_intent import INTENT_KEEP_TMS, classify_group_text self.assertEqual(classify_group_text("第一个", allow_network=False), INTENT_KEEP_TMS) wo, chat = self._open_group() ticket = self.ledger.get_ticket(work_order_no=wo) ticket.quote = { "total": "2287.65", "currency": "CNY", "source_label": "TMS 标准报价", "fee_lines": [ {"name": "操作费", "amount": "680", "currency": "CNY", "unit_price": "680"}, {"name": "文件费", "amount": "450", "currency": "CNY", "unit_price": "450"}, ], } handle_group_collab( InboundMessage( sender_id="prod4", message_id="first-opt-1", content="操作费改为700CNY / 文件费改为500CNY", chat_id=chat, chat_type="group", ), flow=self.flow, ledger=self.ledger, injected_intent="product_quote", ) phase = handle_group_collab( InboundMessage( sender_id="prod4", message_id="first-opt-2", content="第一个", chat_id=chat, chat_type="group", ), flow=self.flow, ledger=self.ledger, ) self.assertEqual(phase, "product_quote_sent") ticket = self.ledger.get_ticket(work_order_no=wo) names = { str(x.get("name") or ""): x for x in (ticket.quote.get("fee_lines") or ticket.quote.get("fee_rows") or []) } self.assertEqual(str(names["操作费"].get("amount") or ""), "700") self.assertEqual(str(names["文件费"].get("amount") or ""), "500") def test_other_product_can_drop_tms(self) -> None: """任一产品都能选；说不用 TMS 就只留刚报的价。""" self.ledger.set_sea_staff( [ { "name": "赵产品", "wecomId": "prod4", "roleCode": "sea", "routes": "海运·全部线路", "status": "active", }, { "name": "钱产品", "wecomId": "prod5", "roleCode": "sea", "routes": "海运·全部线路", "status": "active", }, ] ) wo, chat = self._open_group() ticket = self.ledger.get_ticket(work_order_no=wo) ticket.quote = { "source_label": "TMS 标准报价", "fee_lines": [ {"name": "海运费", "amount": "100.00", "currency": "USD"}, {"name": "报关费", "amount": "250.00", "currency": "CNY"}, ], } handle_group_collab( InboundMessage( sender_id="prod4", message_id="drop-1", content="报关费改为300CNY", chat_id=chat, chat_type="group", ), flow=self.flow, ledger=self.ledger, injected_intent="product_quote", ) phase = handle_group_collab( InboundMessage( sender_id="prod5", message_id="drop-2", content="不要TMS的价，按我刚才报的", chat_id=chat, chat_type="group", ), flow=self.flow, ledger=self.ledger, ) self.assertEqual(phase, "product_quote_sent") ticket = self.ledger.get_ticket(work_order_no=wo) names = { str(x.get("name") or ""): x for x in (ticket.quote.get("fee_lines") or ticket.quote.get("fee_rows") or []) } self.assertEqual(list(names), ["报关费"]) self.assertEqual(str(names["报关费"].get("amount") or ""), "300") def test_parse_group_file_fills_media(self) -> None: from agent.channel.wecom.parse import parse_plain_xml xml = ( "prod41" "filemid-x" "报价.xlsxwr_sea_1" ) msg = parse_plain_xml(xml) self.assertEqual(msg.chat_type, "group") self.assertEqual(msg.chat_id, "wr_sea_1") self.assertEqual(msg.media.get("filename"), "报价.xlsx") self.assertEqual(msg.media.get("media_id"), "mid-x") if __name__ == "__main__": unittest.main()