""" 口述空运抽字段：解析 DeepSeek B JSON，不打外网。 """ from __future__ import annotations import os import sys import unittest from types import SimpleNamespace 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("LEDGER_BACKEND", "memory") from agent.llm.extract_text import extract_inquiry_snapshot from agent.llm.mode_extract_fields import invoke_shell, parse_extract_payload from agent.policy.air_text_flow import AirTextInquiryFlow from agent.ledger.memory_ledger import MemoryLedger ORAL = "上海到洛杉矶空运，3件 180kg，体积2.5，普货纸箱" B_JSON = { "transport_mode": "AIR", "fields": { "起运港": "上海", "目的港": "洛杉矶", "品名": "普货", "件数": "3", "毛重": "180", "体积": "2.5", "包装方式": "纸箱", "报价日期": "", }, } class OralExtractTests(unittest.TestCase): def test_parse_b_json(self) -> None: parsed = parse_extract_payload(B_JSON) self.assertTrue(parsed["ok"]) self.assertEqual(parsed["business_line"], "AIR") self.assertEqual(parsed["facts"]["起运港"], "上海") self.assertNotIn("报价日期", parsed["facts"]) def test_invoke_shell_with_json(self) -> None: out = invoke_shell(payload={"assistant_json": B_JSON}) self.assertTrue(out["ok"]) self.assertEqual(out["facts"]["目的港"], "洛杉矶") def test_invoke_shell_without_json_keeps_legacy(self) -> None: out = invoke_shell(payload={"a": 1}) self.assertEqual(out.get("error"), "extract_fields_shell_not_implemented") def test_oral_snapshot_uses_b_not_regex(self) -> None: snap = extract_inquiry_snapshot(ORAL, b_result=parse_extract_payload(B_JSON)) self.assertEqual(snap["business_line"], "AIR") self.assertEqual(snap["facts"]["件数"], "3件") self.assertEqual(snap["source"], "deepseek_b") def test_oral_without_b_does_not_guess_ports(self) -> None: snap = extract_inquiry_snapshot(ORAL) self.assertEqual(snap["business_line"], "AIR") self.assertFalse(snap["facts"].get("起运港")) def test_assemble_air_query_zhuhai_clark(self) -> None: from agent.schema.tms_air_query import assemble_air_query out = assemble_air_query( { "起运港": "珠海", "目的港": "克拉克", "品名": "茶叶", "件数": "10件", "毛重": "100KGS", "体积": "2CBM", "包装方式": "托盘", "报价日期": "2026-09-12", } ) self.assertTrue(out["ok"]) self.assertEqual(out["facts"]["originCode"], "ZUH") self.assertEqual(out["facts"]["destinationCode"], "CRK") self.assertEqual(out["facts"]["packageType"], "2") self.assertEqual(out["facts"]["件数"], "10件") bulk = assemble_air_query( { "起运港": "珠海", "目的港": "克拉克", "品名": "茶叶", "件数": "10", "毛重": "100", "体积": "2", "包装方式": "散货", "报价日期": "2026-09-12", } ) self.assertTrue(bulk["ok"]) self.assertEqual(bulk["facts"]["包装方式"], "散货") self.assertNotIn("包装方式", bulk["missing"]) def test_fly_route_overwrites_stale_nkg_kul(self) -> None: from agent.schema.field_validate import harvest_oral_measures from agent.schema.tms_air_query import assemble_air_query from agent.policy import inquiry_copy as copy first = harvest_oral_measures( "从南京飞吉隆坡，9月14日茶叶，托盘，4.2cbm，1.5KG", {}, ) self.assertEqual(first["起运港"], "南京") self.assertEqual(first["目的港"], "吉隆坡") assembled = assemble_air_query( {**first, "originCode": "NKG", "destinationCode": "KUL", "件数": "10"} ) card = copy.inquiry_card(work_order_no="WO1", facts=assembled["facts"]) self.assertIn("起运地：南京", card) self.assertIn("目的地：吉隆坡", card) self.assertNotIn("起运地：NKG", card) self.assertNotIn("目的地：KUL", card) second = harvest_oral_measures( "北京飞吉隆坡，9月14日茶叶，4.2cbm，1.5KG", {**assembled["facts"]}, ) again = assemble_air_query(second) self.assertEqual(again["facts"]["起运港"], "北京") self.assertEqual(again["facts"]["originCode"], "PEK") self.assertEqual(again["facts"]["destinationCode"], "KUL") shown = copy.inquiry_card(work_order_no="WO2", facts=again["facts"]) self.assertIn("起运地：北京", shown) self.assertNotIn("起运地：NKG", shown) self.assertNotIn("起运地：南京", shown) def test_oral_pack_and_quote_date_from_current_sentence(self) -> None: from agent.schema.field_validate import harvest_oral_measures first = harvest_oral_measures( "南京飞吉隆坡，9月15日茶叶，托盘，4.2cbm，1.5KG", {}, ) self.assertEqual(first["包装方式"], "托盘") self.assertEqual(first["报价日期"], "2026-09-15") self.assertEqual(first["毛重"], "1.5KG") self.assertEqual(first["体积"], "4.2cbm") second = harvest_oral_measures( "京飞吉隆坡，9月16日茶叶，散货，5.2cbm，2.5KG", first, ) self.assertEqual(second["包装方式"], "散货") self.assertEqual(second["报价日期"], "2026-09-16") self.assertEqual(second["毛重"], "2.5KG") self.assertEqual(second["体积"], "5.2cbm") def test_clarify_replaces_pallet_with_bulk_and_keeps_quote_date(self) -> None: replies: list[str] = [] def reply(text: str, extra=None) -> None: replies.append(text) _ = extra flow = AirTextInquiryFlow(MemoryLedger()) flow.on_text( sender_id="u_date", text="南京飞吉隆坡，9月15日茶叶，托盘，4.2cbm，1.5KG", reply=reply, injected_facts={ "起运港": "南京", "目的港": "吉隆坡", "品名": "茶叶", "包装方式": "托盘", }, injected_mode="AIR", ) self.assertEqual(flow.session_of("u_date").facts["包装方式"], "托盘") self.assertEqual(flow.session_of("u_date").facts["报价日期"], "2026-09-15") flow.on_text( sender_id="u_date", text="京飞吉隆坡，9月16日茶叶，散货，5.2cbm，2.5KG", reply=reply, injected_facts={ "起运港": "南京", "目的港": "吉隆坡", "品名": "茶叶", }, injected_mode="AIR", ) sess = flow.session_of("u_date") self.assertEqual(sess.facts["包装方式"], "散货") self.assertEqual(sess.facts["报价日期"], "2026-09-16") last = replies[-1] self.assertIn("包装类型：散货", last) self.assertIn("报价日期：2026-09-16", last) def test_weight_keeps_user_kg_not_kgs(self) -> None: from agent.policy import inquiry_copy as copy from agent.schema.field_validate import harvest_oral_measures facts = harvest_oral_measures("从南京飞吉隆坡，茶叶，1.5KG，4.2cbm，托盘", {}) self.assertEqual(facts["毛重"], "1.5KG") shown = copy.inquiry_card( work_order_no="WO1", facts={**facts, "起运港": "南京", "目的港": "吉隆坡", "品名": "茶叶", "件数": "10"}, ) self.assertIn("重量：1.5KG", shown) self.assertNotIn("1.5KGS", shown) def test_oral_units_fill_weight_volume_without_b(self) -> None: snap = extract_inquiry_snapshot( "珠海到克拉克空运，茶叶 10件 100KGS 2CBM 托盘" ) self.assertEqual(snap["facts"]["件数"], "10件") self.assertEqual(snap["facts"]["毛重"], "100KGS") self.assertEqual(snap["facts"]["体积"], "2CBM") self.assertFalse(snap["facts"].get("起运港")) def test_labeled_weight_alias(self) -> None: from agent.llm.extract_text import parse_labeled_facts from agent.schema.field_validate import normalize_facts facts = parse_labeled_facts("重量(KG)：180\n体积(CBM)：2.5") norm = normalize_facts(facts) self.assertEqual(norm["毛重"], "180") self.assertEqual(norm["体积"], "2.5") def test_chat_fn_fixture_drives_flow(self) -> None: def chat_fn(**kwargs): return SimpleNamespace( ok=True, content="", error="", tool_calls=[ { "function": { "name": "extract_inquiry_fields", "arguments": __import__("json").dumps(B_JSON, ensure_ascii=False), } } ], ) replies: list[str] = [] def reply(text: str, extra=None) -> None: replies.append(text) _ = extra flow = AirTextInquiryFlow(MemoryLedger()) phase = flow.on_text( sender_id="oral1", text=ORAL, reply=reply, chat_fn=chat_fn, ) self.assertEqual(phase, "wait_collab") joined = "\n".join(replies) self.assertIn("工单号：", joined) self.assertIn("起运地：上海", joined) self.assertIn("目的地：洛杉矶", joined) self.assertNotIn("【询价卡片】", joined) if __name__ == "__main__": unittest.main()