qizhong/inquiry_robot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

补齐成交跟进：按线路推成交通知、收下未成交原因、协商中按工作日再问。

Browse Source 
成交走询价消息服务；陆运/海运私聊下一句能记下原因；协商中回执写明顺延三个工作日和下次 9：00。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
jillion886
2026-09-19 16:08:48 +08:00
co-authored by Cursor
parent 9bb69ae72e
commit 1a2e1d6cb2
31 changed files with 1638 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
inquiry-agent/tests/test_deal_followup.py
+144
View File
@@ -0,0 +1,144 @@
"""
成交跟进：工作日顺延、线路对负责人、未成交收下原因（不打外网）。
"""
from __future__ import annotations
import unittest
from datetime import date, datetime
from zoneinfo import ZoneInfo
from agent.policy.deal_followup import (
SHANGHAI,
air_route_keys,
leader_matches,
next_followup_at,
shift_workdays,
)
class TestWorkdays(unittest.TestCase):
def test_skip_weekend(self) -> None:
# 2026-09-18 周五，顺延 3 个工作日 → 21、22、23，第四个工作日 24
self.assertEqual(shift_workdays(date(2026, 9, 18), 3), date(2026, 9, 23))
self.assertEqual(shift_workdays(date(2026, 9, 18), 4), date(2026, 9, 24))
def test_skip_national_day(self) -> None:
# 2026-09-30 周三，国庆休 → 第四个工作日落到节后
fourth = shift_workdays(date(2026, 9, 30), 4)
self.assertGreaterEqual(fourth, date(2026, 10, 9))
self.assertNotEqual(fourth.weekday(), 5)
self.assertNotEqual(fourth.weekday(), 6)
def test_next_followup_is_9am(self) -> None:
when = next_followup_at(datetime(2026, 9, 18, 15, 30, tzinfo=SHANGHAI))
self.assertEqual(when.hour, 9)
self.assertEqual(when.minute, 0)
self.assertEqual(when.tzinfo, SHANGHAI)
self.assertEqual(when.date(), date(2026, 9, 24))
class TestAirRouteKeys(unittest.TestCase):
def test_pair_and_dest(self) -> None:
keys = air_route_keys({"起运港": "SZX", "目的港": "KUL"})
self.assertIn("SZX-KUL", keys)
self.assertIn("KUL", keys)
def test_chinese_city_projects_iata(self) -> None:
keys = air_route_keys({"起运港": "深圳", "目的港": "吉隆坡"})
self.assertIn("SZX-KUL", keys)
self.assertIn("KUL", keys)
class TestLeaderMatch(unittest.TestCase):
def test_sea_all_line(self) -> None:
self.assertTrue(
leader_matches(
role_code="sea",
routes="海运全线",
is_leader=True,
business_line="SEA",
facts={},
)
)
self.assertFalse(
leader_matches(
role_code="sea",
routes="海运全线",
is_leader=False,
business_line="SEA",
facts={},
)
)
def test_land_route_category(self) -> None:
self.assertTrue(
leader_matches(
role_code="land",
routes="国内长途/零担,东南亚",
is_leader=True,
business_line="LAND",
facts={"线路类别": "国内长途/零担"},
)
)
self.assertFalse(
leader_matches(
role_code="land",
routes="东南亚",
is_leader=True,
business_line="LAND",
facts={"线路类别": "国内长途/零担"},
)
)
def test_air_pair_dest_and_all(self) -> None:
facts = {"起运港": "SZX", "目的港": "KUL"}
self.assertTrue(
leader_matches(
role_code="air",
routes="SZX-KUL",
is_leader=True,
business_line="AIR",
facts=facts,
)
)
self.assertTrue(
leader_matches(
role_code="air",
routes="KUL",
is_leader=True,
business_line="AIR",
facts=facts,
)
)
self.assertTrue(
leader_matches(
role_code="air",
routes="全线",
is_leader=True,
business_line="AIR",
facts=facts,
)
)
self.assertFalse(
leader_matches(
role_code="air",
routes="外采-空运",
is_leader=True,
business_line="AIR",
facts=facts,
)
)
self.assertTrue(
leader_matches(
role_code="air",
routes="外采-空运",
is_leader=True,
business_line="AIR",
facts={**facts, "外采": "1"},
)
)
if __name__ == "__main__":
unittest.main()
inquiry-agent/tests/test_deal_outcome.py
+247
View File
@@ -0,0 +1,247 @@
"""
成交三态：入账、推负责人、收下未成交原因、协商中不停会话。
"""
from __future__ import annotations
import unittest
from agent.ledger.memory_ledger import MemoryLedger
from agent.policy import inquiry_copy as copy
from agent.policy.air_text_flow import AirTextInquiryFlow, FlowSession
from agent.policy.deal_outcome import capture_lost_reason, finish_deal_outcome
from agent.policy.sea_text_flow import SeaTextInquiryFlow
class TestDealOutcome(unittest.TestCase):
def setUp(self) -> None:
self.ledger = MemoryLedger()
created = self.ledger.create_ticket(
sender_id="sales1",
business_line="SEA",
facts={"起运港": "上海", "目的港": "洛杉矶", "品名": "衣服"},
)
self.wo = str(created.get("work_order_no") or "")
self.ledger.transition(work_order_no=self.wo, to_status="已报价")
self.ledger.upsert_staff(
{
"name": "海运负责人",
"wecomId": "lead-sea",
"roleCode": "sea",
"routes": "海运全线",
"isLeader": 1,
"status": "active",
}
)
def test_deal_notifies_leaders(self) -> None:
queued: list[dict] = []
def enqueue(*, touser, content, dedupe_key, payload):
queued.append(
{
"touser": touser,
"content": content,
"dedupe_key": dedupe_key,
"payload": payload,
}
)
return True, "ok"
out = finish_deal_outcome(
ledger=self.ledger,
work_order_no=self.wo,
outcome="已成交",
facts={"起运港": "上海", "目的港": "洛杉矶"},
quote={"total": "1000", "currency": "USD"},
business_line="SEA",
sales_name="吴吉林",
enqueue=enqueue,
notify_ready=True,
)
self.assertTrue(out.get("ok"))
self.assertEqual(out.get("phase"), "done")
self.assertIn("海运负责人", str(out.get("notify_summary") or ""))
self.assertEqual(queued[0]["touser"], "lead-sea")
self.assertEqual(queued[0]["payload"].get("wecom_app"), "notify")
ticket = self.ledger.get_ticket(work_order_no=self.wo)
self.assertEqual(ticket.status, "已成交")
self.assertEqual(ticket.next_followup_at, "")
def test_deal_notify_card_omits_sales_label(self) -> None:
"""成交通知不要「销售」二字：标题副标和正文都不写。"""
text = copy.deal_notify_card(
work_order_no="WO202609180008",
sales_name="吴吉林",
facts={"起运地": "泉州", "目的地": "南京", "线路类别": "国内城配/拖车/打包"},
quote={},
business_line="LAND",
)
self.assertNotIn("销售", text)
payload = copy.deal_notify_wecom_payload(
work_order_no="WO202609180008",
sales_name="吴吉林",
facts={"起运地": "泉州", "目的地": "南京", "线路类别": "国内城配/拖车/打包"},
quote={},
business_line="LAND",
)
title = (payload.get("template_card") or {}).get("main_title") or {}
blob = str(title.get("title") or "") + str(title.get("desc") or "")
self.assertNotIn("销售", blob)
self.assertNotIn("desc", title)
def test_deal_without_leader_still_closes(self) -> None:
self.ledger.upsert_staff(
{
"name": "不是负责人",
"wecomId": "lead-sea",
"roleCode": "sea",
"routes": "海运全线",
"isLeader": 0,
"status": "active",
}
)
out = finish_deal_outcome(
ledger=self.ledger,
work_order_no=self.wo,
outcome="已成交",
facts={},
quote={},
business_line="SEA",
notify_ready=True,
)
self.assertEqual(out.get("notify_summary"), "无匹配负责人")
self.assertEqual(self.ledger.get_ticket(work_order_no=self.wo).status, "已成交")
def test_lost_waits_for_reason(self) -> None:
out = finish_deal_outcome(
ledger=self.ledger,
work_order_no=self.wo,
outcome="未成交",
facts={},
quote={},
business_line="SEA",
)
self.assertEqual(out.get("phase"), "wait_lost_reason")
ticket = self.ledger.get_ticket(work_order_no=self.wo)
self.assertEqual(ticket.status, "未成交")
self.assertEqual(ticket.lost_reason, "")
view = self.ledger.get_for_agent(work_order_no=self.wo, sender_id="sales1")
self.assertEqual(view.get("wait_phase"), "wait_lost_reason")
captured = capture_lost_reason(
ledger=self.ledger, work_order_no=self.wo, reason="客人比价了"
)
self.assertTrue(captured.get("ok"))
ticket = self.ledger.get_ticket(work_order_no=self.wo)
self.assertEqual(ticket.lost_reason, "客人比价了")
view = self.ledger.get_for_agent(work_order_no=self.wo, sender_id="sales1")
self.assertEqual(view.get("wait_phase"), "done")
def test_lost_reason_ack_repeats_user_text(self) -> None:
self.assertEqual(copy.lost_reason_ack("客户不要了"), "已记下未成交原因：客户不要了。")
self.assertEqual(copy.lost_reason_ack("价格高。"), "已记下未成交原因：价格高。")
def test_negotiate_ack_names_followup_day(self) -> None:
text = copy.deal_ack("协商中", "WO202609190007", "2026-09-24 09:00:00")
self.assertEqual(
text,
"已记录为「协商中」。已顺延三个工作日，9月24日的9：00再次发起成交状态询问。",
)
self.assertNotIn("跟进周期已顺延", text)
def test_negotiate_keeps_wait_deal(self) -> None:
out = finish_deal_outcome(
ledger=self.ledger,
work_order_no=self.wo,
outcome="协商中",
facts={},
quote={},
business_line="SEA",
)
self.assertEqual(out.get("phase"), "wait_deal")
ticket = self.ledger.get_ticket(work_order_no=self.wo)
self.assertEqual(ticket.status, "协商中")
self.assertTrue(ticket.next_followup_at)
view = self.ledger.get_for_agent(work_order_no=self.wo, sender_id="sales1")
self.assertEqual(view.get("wait_phase"), "wait_deal")
again = finish_deal_outcome(
ledger=self.ledger,
work_order_no=self.wo,
outcome="协商中",
facts={},
quote={},
business_line="SEA",
)
self.assertEqual(again.get("phase"), "wait_deal")
self.assertEqual(self.ledger.get_ticket(work_order_no=self.wo).status, "协商中")
self.assertTrue(again.get("next_followup_at"))
def test_private_next_text_is_lost_reason(self) -> None:
flow = AirTextInquiryFlow(self.ledger)
sess = FlowSession(
sender_id="sales1",
thread_id="sales1",
phase="wait_deal",
business_line="SEA",
work_order_no=self.wo,
allowed=(copy.BTN_DEAL, copy.BTN_LOST, copy.BTN_NEGOTIATE),
)
flow._save(sess)
replies: list[str] = []
phase = flow.on_button(
sender_id="sales1",
action=copy.BTN_LOST,
reply=lambda t, extra=None: replies.append(t),
)
self.assertEqual(phase, "wait_lost_reason")
phase = flow.on_text(
sender_id="sales1",
text="价格高",
reply=lambda t, extra=None: replies.append(t),
)
self.assertEqual(phase, "done")
self.assertEqual(self.ledger.get_ticket(work_order_no=self.wo).lost_reason, "价格高")
self.assertTrue(any(t == "已记下未成交原因：价格高。" for t in replies))
def test_sea_private_next_text_is_lost_reason(self) -> None:
flow = SeaTextInquiryFlow(self.ledger)
flow._save(
FlowSession(
sender_id="sales1",
thread_id="sales1",
phase="wait_deal",
business_line="SEA",
work_order_no=self.wo,
allowed=(copy.BTN_DEAL, copy.BTN_LOST, copy.BTN_NEGOTIATE),
)
)
replies: list[str] = []
phase = flow.on_button(
sender_id="sales1",
action=copy.BTN_LOST,
reply=lambda t, extra=None: replies.append(t),
)
self.assertEqual(phase, "wait_lost_reason")
phase = flow.on_text(
sender_id="sales1",
text="客人比价了",
reply=lambda t, extra=None: replies.append(t),
)
self.assertEqual(phase, "done")
self.assertEqual(self.ledger.get_ticket(work_order_no=self.wo).lost_reason, "客人比价了")
def test_claim_due_negotiate_clears_schedule(self) -> None:
finish_deal_outcome(
ledger=self.ledger,
work_order_no=self.wo,
outcome="协商中",
facts={},
quote={},
business_line="SEA",
)
ticket = self.ledger.get_ticket(work_order_no=self.wo)
ticket.next_followup_at = "2020-01-01 09:00:00"
rows = self.ledger.claim_due_negotiate()
self.assertEqual(len(rows), 1)
self.assertEqual(rows[0]["work_order_no"], self.wo)
self.assertEqual(self.ledger.get_ticket(work_order_no=self.wo).next_followup_at, "")
inquiry-agent/tests/test_land_text_inquiry.py
+42
View File
@@ -26,6 +26,7 @@ os.environ.setdefault("LEDGER_BACKEND", "memory")
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.land_text_flow import LandTextInquiryFlow
@@ -623,5 +624,46 @@ class LandTextPullTests(unittest.TestCase):
self.assertTrue(is_product(ticket, "u1", flow=self.flow))
class LandLostReasonTests(unittest.TestCase):
"""陆运私聊未成交后，下一句任意文字要记下原因，不能当新询价丢掉。"""
def test_land_private_next_text_is_lost_reason(self) -> None:
ledger = MemoryLedger()
created = ledger.create_ticket(
sender_id="WuJiLin",
business_line="LAND",
facts={"起运地": "泉州", "目的地": "南京", "线路类别": "国内城配/拖车/打包"},
)
wo = str(created.get("work_order_no") or "")
ledger.transition(work_order_no=wo, to_status="已报价")
flow = LandTextInquiryFlow(ledger)
flow._save(
FlowSession(
sender_id="WuJiLin",
thread_id="WuJiLin",
phase="wait_deal",
business_line="LAND",
work_order_no=wo,
allowed=(copy.BTN_DEAL, copy.BTN_LOST, copy.BTN_NEGOTIATE),
)
)
replies: list[str] = []
phase = flow.on_button(
sender_id="WuJiLin",
action=copy.BTN_LOST,
reply=lambda t, extra=None: replies.append(t),
)
self.assertEqual(phase, "wait_lost_reason")
self.assertEqual(ledger.get_ticket(work_order_no=wo).status, "未成交")
phase = flow.on_text(
sender_id="WuJiLin",
text="价格高",
reply=lambda t, extra=None: replies.append(t),
)
self.assertEqual(phase, "done")
self.assertEqual(ledger.get_ticket(work_order_no=wo).lost_reason, "价格高")
self.assertTrue(any(t == "已记下未成交原因：价格高。" for t in replies))
if __name__ == "__main__":
unittest.main()
inquiry-agent/tests/test_sea_text_inquiry.py
+28 -1
View File
@@ -1522,9 +1522,36 @@ class SeaGroupOpsTests(unittest.TestCase):
flow=self.flow,
ledger=self.ledger,
)
self.assertEqual(phase, "done")
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_can_close_deal_by_text(self) -> None:
"""群里销售说成交，要入账并回话，不能装没听见。相近说法也认。"""
wo, chat = self._open_group()