私聊建单前拦住枪支弹药爆炸物和毒品，避免真家伙进入报价。
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Cursor
@@ -0,0 +1,193 @@
|
||||
"""
|
||||
私聊建单前违禁品：词表、改口、不建单。
|
||||
|
||||
不连企微、TMS、主账。群聊不走这条拦截。
|
||||
"""
|
||||
|
||||
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.ledger.memory_ledger import MemoryLedger
|
||||
from agent.policy.sensitive_cargo import BLOCK_REPLY, contains_blocked, judge
|
||||
from agent.policy.sea_text_flow import SeaTextInquiryFlow, reset_sea_text_flow_for_test
|
||||
|
||||
COMPLETE_SEA = {
|
||||
"起运港": "上海",
|
||||
"目的港": "洛杉矶",
|
||||
"品名": "普货",
|
||||
"货量": "20吨",
|
||||
"整柜或拼柜": "整柜",
|
||||
"箱型箱量": "1x40HQ",
|
||||
"贸易条款": "FOB",
|
||||
"运输分类": "Port to Port",
|
||||
}
|
||||
|
||||
|
||||
class WordListTests(unittest.TestCase):
|
||||
def test_real_weapons_and_drugs_block(self) -> None:
|
||||
for word in ("炸弹", "枪支", "手枪", "子弹", "海洛因", "火炮"):
|
||||
self.assertTrue(contains_blocked(f"上海到洛杉矶，品名{word}"), word)
|
||||
|
||||
def test_toys_and_fireworks_pass(self) -> None:
|
||||
for word in ("玩具枪", "水枪", "仿真枪", "气枪", "烟花", "礼炮", "鞭炮"):
|
||||
self.assertFalse(contains_blocked(f"品名{word}"), word)
|
||||
|
||||
def test_toy_prefix_does_not_hide_real_gun(self) -> None:
|
||||
self.assertTrue(contains_blocked("玩具枪支"))
|
||||
|
||||
def test_ordinary_dangerous_goods_pass(self) -> None:
|
||||
self.assertFalse(contains_blocked("这票是危险品，含电，锂电池"))
|
||||
|
||||
|
||||
class JudgeTests(unittest.TestCase):
|
||||
def test_stray_sentence_sticks_until_cargo_rename(self) -> None:
|
||||
facts = {"品名": "服装", "起运港": "上海"}
|
||||
blocked, hold, _edits = judge(text="客户提到有炸弹", facts=dict(facts), hold="")
|
||||
self.assertTrue(blocked)
|
||||
self.assertIn("炸弹", hold)
|
||||
still, hold2, _edits = judge(text="重量100KGS", facts=dict(facts), hold=hold)
|
||||
self.assertTrue(still)
|
||||
self.assertEqual(hold2, hold)
|
||||
released, hold3, edits = judge(text="品名改为服装", facts=dict(facts), hold=hold)
|
||||
self.assertFalse(released)
|
||||
self.assertEqual(hold3, "")
|
||||
self.assertEqual(edits.get("品名"), "服装")
|
||||
|
||||
def test_rename_sentence_that_still_says_the_word_stays_blocked(self) -> None:
|
||||
blocked, hold, _edits = judge(text="不是炸弹，品名服装", facts={"品名": "服装"}, hold="")
|
||||
self.assertTrue(blocked)
|
||||
self.assertIn("炸弹", hold)
|
||||
|
||||
def test_origin_and_remark_remain_after_cargo_rename(self) -> None:
|
||||
facts = {"品名": "炸弹", "起运港": "上海", "备注": "内有枪支"}
|
||||
blocked, hold, _edits = judge(text="品名改为服装", facts=facts, hold="品名炸弹")
|
||||
self.assertTrue(blocked)
|
||||
self.assertEqual(hold, "")
|
||||
self.assertEqual(facts["品名"], "服装")
|
||||
self.assertIn("枪支", facts["备注"])
|
||||
|
||||
def test_remark_can_be_rewritten(self) -> None:
|
||||
facts = {"品名": "服装", "备注": "枪支"}
|
||||
blocked, hold, _edits = judge(text="备注改为无", facts=facts, hold="")
|
||||
self.assertFalse(blocked)
|
||||
self.assertEqual(facts["备注"], "无")
|
||||
self.assertEqual(hold, "")
|
||||
|
||||
|
||||
class SeaPrivateFlowTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
reset_sea_text_flow_for_test()
|
||||
self.ledger = MemoryLedger()
|
||||
self.flow = SeaTextInquiryFlow(self.ledger)
|
||||
self.replies: list[str] = []
|
||||
|
||||
def _reply(self, text: str, extra=None) -> None:
|
||||
_ = extra
|
||||
self.replies.append(text)
|
||||
|
||||
def test_private_text_blocks_before_ticket(self) -> None:
|
||||
facts = dict(COMPLETE_SEA)
|
||||
facts["品名"] = "炸弹"
|
||||
phase = self.flow.on_text(
|
||||
sender_id="s-bomb",
|
||||
text="海运 上海到洛杉矶 品名炸弹",
|
||||
reply=self._reply,
|
||||
injected_facts=facts,
|
||||
injected_mode="SEA",
|
||||
)
|
||||
self.assertEqual(phase, "sensitive_block")
|
||||
self.assertEqual(self.replies, [BLOCK_REPLY])
|
||||
self.assertEqual(self.ledger._tickets, {})
|
||||
sess = self.flow.session_of("s-bomb")
|
||||
self.assertFalse((sess.work_order_no or "").strip())
|
||||
|
||||
def test_file_facts_block_even_if_caption_is_plain(self) -> None:
|
||||
facts = dict(COMPLETE_SEA)
|
||||
facts["品名"] = "枪支"
|
||||
phase = self.flow.on_text(
|
||||
sender_id="s-file",
|
||||
text="海运报价单",
|
||||
reply=self._reply,
|
||||
injected_facts=facts,
|
||||
injected_mode="SEA",
|
||||
)
|
||||
self.assertEqual(phase, "sensitive_block")
|
||||
self.assertEqual(self.ledger._tickets, {})
|
||||
|
||||
def test_rename_then_confirm_creates_ticket(self) -> None:
|
||||
facts = dict(COMPLETE_SEA)
|
||||
facts["品名"] = "炸弹"
|
||||
self.flow.on_text(
|
||||
sender_id="s-fix",
|
||||
text="海运 品名炸弹 上海到洛杉矶",
|
||||
reply=self._reply,
|
||||
injected_facts=facts,
|
||||
injected_mode="SEA",
|
||||
)
|
||||
self.replies.clear()
|
||||
fixed = dict(COMPLETE_SEA)
|
||||
phase = self.flow.on_text(
|
||||
sender_id="s-fix",
|
||||
text="品名改为服装",
|
||||
reply=self._reply,
|
||||
injected_facts=fixed,
|
||||
injected_mode="SEA",
|
||||
)
|
||||
self.assertNotEqual(phase, "sensitive_block")
|
||||
self.assertNotIn(BLOCK_REPLY, self.replies)
|
||||
self.replies.clear()
|
||||
created = self.flow.on_text(sender_id="s-fix", text="确定", reply=self._reply)
|
||||
self.assertNotEqual(created, "sensitive_block")
|
||||
self.assertTrue(self.ledger._tickets)
|
||||
joined = "\n".join(self.replies)
|
||||
self.assertNotIn(BLOCK_REPLY, joined)
|
||||
|
||||
def test_weight_alone_does_not_clear(self) -> None:
|
||||
facts = dict(COMPLETE_SEA)
|
||||
facts["品名"] = "弹药"
|
||||
self.flow.on_text(
|
||||
sender_id="s-hold",
|
||||
text="海运 品名弹药",
|
||||
reply=self._reply,
|
||||
injected_facts=facts,
|
||||
injected_mode="SEA",
|
||||
)
|
||||
self.replies.clear()
|
||||
phase = self.flow.on_text(
|
||||
sender_id="s-hold",
|
||||
text="重量 20吨",
|
||||
reply=self._reply,
|
||||
injected_facts=facts,
|
||||
injected_mode="SEA",
|
||||
)
|
||||
self.assertEqual(phase, "sensitive_block")
|
||||
self.assertEqual(self.replies, [BLOCK_REPLY])
|
||||
self.assertEqual(self.ledger._tickets, {})
|
||||
|
||||
def test_toy_gun_can_continue(self) -> None:
|
||||
facts = dict(COMPLETE_SEA)
|
||||
facts["品名"] = "玩具枪"
|
||||
phase = self.flow.on_text(
|
||||
sender_id="s-toy",
|
||||
text="海运 品名玩具枪 上海到洛杉矶",
|
||||
reply=self._reply,
|
||||
injected_facts=facts,
|
||||
injected_mode="SEA",
|
||||
)
|
||||
self.assertEqual(phase, "wait_confirm")
|
||||
self.assertNotIn(BLOCK_REPLY, self.replies)
|
||||
Reference in New Issue
Block a user