diff --git a/inquiry-agent/agent/schema/sea_options.py b/inquiry-agent/agent/schema/sea_options.py new file mode 100644 index 0000000..f121b3e --- /dev/null +++ b/inquiry-agent/agent/schema/sea_options.py @@ -0,0 +1,64 @@ +""" +海运运输分类:只认 Port to Port、其他。 + +本文件职责:从原话收成标准词、生成补问清单。 +禁止:发明第三项;把分类带去 TMS;改六态。 +线程:纯函数,可在 Worker 槽调用。 +""" + +from __future__ import annotations + +SEA_CLASSES = ("Port to Port", "其他") + +# 长词优先,避免 P2P 误伤其它词;门到门不在名单里。 +_ALIASES = ( + ("port to port", "Port to Port"), + ("港到港", "Port to Port"), + ("p2p", "Port to Port"), + ("其它", "其他"), + ("其他", "其他"), +) + + +def canonicalize_sea_class(raw: str) -> str: + """ + 把销售原话收成 Port to Port 或 其他;对不上返回空串。 + + 「门到门」不算。先匹配更长别名。副作用:无。 + """ + text = (raw or "").strip() + if not text: + return "" + lowered = text.lower() + for needle, canon in _ALIASES: + if needle in lowered or needle in text: + return canon + stripped = text.replace("运输分类", "").replace(":", "").replace(":", "").strip() + if stripped in SEA_CLASSES: + return stripped + return "" + + +def harvest_sea_class(text: str, current: dict[str, str] | None = None) -> dict[str, str]: + """ + 当前句能认则覆盖;认不出则保留已有标准值。 + + 参数:本轮原文、已有 facts。返回合并后的 facts 副本。 + 非法旧值会丢掉,避免「门到门」一直占着必填位。 + """ + merged = dict(current or {}) + found = canonicalize_sea_class(text or "") + if found: + merged["运输分类"] = found + return merged + kept = canonicalize_sea_class(str(merged.get("运输分类") or "")) + if kept: + merged["运输分类"] = kept + else: + merged.pop("运输分类", None) + return merged + + +def sea_class_prompt() -> str: + """缺运输分类时只列两项,不再追问「其他是什么」。""" + return "运输分类(必填):请回复 Port to Port 或 其他。" diff --git a/inquiry-agent/tests/test_sea_options.py b/inquiry-agent/tests/test_sea_options.py new file mode 100644 index 0000000..cb2abec --- /dev/null +++ b/inquiry-agent/tests/test_sea_options.py @@ -0,0 +1,52 @@ +""" +海运运输分类:只认 Port to Port、其他。 +""" + +from __future__ import annotations + +import os +import sys +import unittest + +ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) +if ROOT not in sys.path: + sys.path.insert(0, ROOT) + +from agent.schema.sea_options import ( + canonicalize_sea_class, + harvest_sea_class, + sea_class_prompt, +) + + +class SeaOptionsTest(unittest.TestCase): + def test_port_to_port_aliases(self): + self.assertEqual(canonicalize_sea_class("Port to Port"), "Port to Port") + self.assertEqual(canonicalize_sea_class("port to port"), "Port to Port") + self.assertEqual(canonicalize_sea_class("P2P"), "Port to Port") + self.assertEqual(canonicalize_sea_class("港到港"), "Port to Port") + + def test_other_alias(self): + self.assertEqual(canonicalize_sea_class("其他"), "其他") + self.assertEqual(canonicalize_sea_class("其它"), "其他") + + def test_door_to_door_rejected(self): + self.assertEqual(canonicalize_sea_class("门到门"), "") + + def test_harvest_keeps_current_when_text_has_none(self): + got = harvest_sea_class("上海到洛杉矶整柜", {"运输分类": "Port to Port"}) + self.assertEqual(got["运输分类"], "Port to Port") + + def test_harvest_overrides_from_text(self): + got = harvest_sea_class("运输分类:其他", {"运输分类": "Port to Port"}) + self.assertEqual(got["运输分类"], "其他") + + def test_prompt_lists_two_options_only(self): + text = sea_class_prompt() + self.assertIn("Port to Port", text) + self.assertIn("其他", text) + self.assertNotIn("门到门", text) + + +if __name__ == "__main__": + unittest.main()