Files
inquiry_robot/inquiry-agent/tests/test_quote_adjust.py
T

497 lines
24 KiB
Python

"""
调价规则单测:比例、改到金额、币种合计、盖价、利润点。
不连企微 / 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)
from agent.policy import inquiry_copy as copy
from agent.schema.quote_adjust import (
apply_adjust,
effective_quote,
extract_adjust,
private_session_can_adjust,
total_price_label,
)
def _quote(*rows, total="", currency="CNY"):
lines = [{"name": n, "amount": a, "currency": c} for n, a, c in rows]
return {
"total": total,
"currency": currency,
"fee_lines": lines,
"fee_rows": lines,
}
class QuoteAdjustTests(unittest.TestCase):
def test_percent_up_single_currency_updates_total(self) -> None:
src = _quote(("提货费", "1000", "CNY"), ("送货费", "200", "CNY"), total="1200")
ext = extract_adjust("提货费加 10%", fee_names=["提货费", "送货费"])
got = apply_adjust(src, ext, business_line="SEA", tms_cost="1200", tms_sale="1200")
self.assertTrue(got.ok)
self.assertEqual(got.code, "applied")
self.assertEqual(got.quote["fee_rows"][0]["amount"], "1100.00")
self.assertEqual(str(got.quote["total"]), "1300.00")
self.assertFalse(got.currency_hint)
def test_percent_down_and_target_in_one_sentence(self) -> None:
src = _quote(("提货费", "1000", "CNY"), ("送货费", "200", "CNY"), total="1200")
ext = extract_adjust("提货费减 5%,送货费提升到 180 元", fee_names=["提货费", "送货费"])
got = apply_adjust(src, ext, business_line="LAND", tms_cost="1200")
self.assertTrue(got.ok)
names = {r["name"]: r["amount"] for r in got.quote["fee_rows"]}
self.assertEqual(names["提货费"], "950.00")
self.assertEqual(names["送货费"], "180.00")
def test_unspecified_quote_does_not_change(self) -> None:
src = _quote(("提货费", "1000", "CNY"), total="1000")
ext = extract_adjust("报价加 10%", fee_names=["提货费"])
got = apply_adjust(src, ext, business_line="AIR")
self.assertFalse(got.ok)
self.assertEqual(got.code, "need_fee")
self.assertEqual(src["fee_rows"][0]["amount"], "1000")
def test_single_line_same_currency_total_moves_line(self) -> None:
"""明细只有一项且币种与总价一致:改总价也改明细,不看出价是否相等。"""
src = _quote(("运输费", "110", "CNY"), total="200")
ext = extract_adjust("总销售价改为180CNY", fee_names=["运输费"], allow_model=False)
got = apply_adjust(src, ext, business_line="LAND")
self.assertTrue(got.ok)
self.assertEqual(str(got.quote["total"]), "180.00")
self.assertEqual(got.quote["fee_rows"][0]["amount"], "180.00")
self.assertFalse(got.lines_hint)
def test_single_line_same_currency_line_moves_total(self) -> None:
src = _quote(("运输费", "110", "CNY"), total="200")
ext = extract_adjust("运输费提升到 180", fee_names=["运输费"], allow_model=False)
got = apply_adjust(src, ext, business_line="LAND")
self.assertTrue(got.ok)
self.assertEqual(got.quote["fee_rows"][0]["amount"], "180.00")
self.assertEqual(str(got.quote["total"]), "180.00")
def test_single_line_other_currency_does_not_sync(self) -> None:
src = _quote(("运输费", "110", "USD"), total="200", currency="CNY")
ext = extract_adjust("总销售价改为180CNY", fee_names=["运输费"], allow_model=False)
got = apply_adjust(src, ext, business_line="LAND")
self.assertTrue(got.ok)
self.assertEqual(str(got.quote["total"]), "180.00")
self.assertEqual(got.quote["fee_rows"][0]["amount"], "110")
self.assertTrue(got.currency_hint)
self.assertFalse(got.lines_hint)
def test_multi_line_total_does_not_allocate(self) -> None:
src = _quote(("提货费", "1000", "CNY"), ("送货费", "200", "CNY"), total="1200")
ext = extract_adjust("总销售价改为180CNY", fee_names=["提货费", "送货费"], allow_model=False)
got = apply_adjust(src, ext, business_line="LAND")
self.assertTrue(got.ok)
self.assertEqual(str(got.quote["total"]), "180.00")
names = {r["name"]: r["amount"] for r in got.quote["fee_rows"]}
self.assertEqual(names["提货费"], "1000")
self.assertEqual(names["送货费"], "200")
self.assertTrue(got.lines_hint)
def test_multi_currency_named_usd_does_not_change_cny_total(self) -> None:
src = _quote(
("提货费", "100", "USD"),
("空运费", "2000", "CNY"),
total="2000",
currency="CNY",
)
ext = extract_adjust("提货费加 10%", fee_names=["提货费", "空运费"])
got = apply_adjust(src, ext, business_line="AIR", tms_sale="2000")
self.assertTrue(got.ok)
names = {r["name"]: r["amount"] for r in got.quote["fee_rows"]}
self.assertEqual(names["提货费"], "110.00")
self.assertEqual(names["空运费"], "2000")
self.assertEqual(str(got.quote["total"]), "2000")
self.assertTrue(got.currency_hint)
def test_multi_currency_line_does_not_change_total(self) -> None:
src = _quote(
("提货费", "100", "USD"),
("空运费", "2000", "CNY"),
total="2000",
currency="CNY",
)
ext = extract_adjust("空运费加 10%", fee_names=["提货费", "空运费"], allow_model=False)
got = apply_adjust(src, ext, business_line="AIR")
self.assertTrue(got.ok)
names = {r["name"]: r["amount"] for r in got.quote["fee_rows"]}
self.assertEqual(names["空运费"], "2200.00")
self.assertEqual(names["提货费"], "100")
self.assertEqual(str(got.quote["total"]), "2000")
self.assertTrue(got.currency_hint)
def test_multi_currency_total_only_changes_total(self) -> None:
src = _quote(
("提货费", "100", "USD"),
("空运费", "2000", "CNY"),
total="2000",
currency="CNY",
)
ext = extract_adjust("总销售价改为180CNY", fee_names=["提货费", "空运费"], allow_model=False)
got = apply_adjust(src, ext, business_line="AIR")
self.assertTrue(got.ok)
names = {r["name"]: r["amount"] for r in got.quote["fee_rows"]}
self.assertEqual(names["提货费"], "100")
self.assertEqual(names["空运费"], "2000")
self.assertEqual(str(got.quote["total"]), "180.00")
self.assertTrue(got.currency_hint)
self.assertFalse(got.lines_hint)
def test_partial_unmatched_still_applies_matched(self) -> None:
src = _quote(("卸货费", "300", "CNY"), total="300")
ext = extract_adjust("卸货费加 10%,报关费减 5%", fee_names=["卸货费"])
got = apply_adjust(src, ext, business_line="SEA")
self.assertTrue(got.ok)
self.assertIn("报关费", got.unmatched)
self.assertEqual(got.quote["fee_rows"][0]["amount"], "330.00")
def test_none_matched(self) -> None:
src = _quote(("卸货费", "300", "CNY"), total="300")
ext = extract_adjust("报关费减 5%", fee_names=["卸货费"])
got = apply_adjust(src, ext, business_line="SEA")
self.assertFalse(got.ok)
self.assertEqual(got.code, "none_matched")
def test_overlay_manual_wins(self) -> None:
tms = _quote(("运输费", "1000", "CNY"), ("卸货费", "200", "CNY"), total="1200")
manual = _quote(("卸货费", "300", "CNY"), ("配送费", "20", "CNY"), total="320")
merged = effective_quote(tms, manual)
names = {r["name"]: r["amount"] for r in merged["fee_rows"]}
self.assertEqual(names["运输费"], "1000")
self.assertEqual(names["卸货费"], "300")
self.assertEqual(names["配送费"], "20")
ext = extract_adjust("卸货费加 10%", fee_names=list(names))
got = apply_adjust(merged, ext, business_line="SEA", tms_cost="1200")
after = {r["name"]: r["amount"] for r in got.quote["fee_rows"]}
self.assertEqual(after["卸货费"], "330.00")
self.assertEqual(after["运输费"], "1000")
self.assertEqual(after["配送费"], "20")
def test_profit_sea_uses_cost(self) -> None:
src = _quote(("运输费", "1000", "CNY"), total="1000")
ext = extract_adjust("运输费提升到 1300 元", fee_names=["运输费"])
got = apply_adjust(src, ext, business_line="SEA", tms_cost="1000", tms_sale="1100")
self.assertAlmostEqual(got.profit_point or 0, 30.0)
def test_profit_air_uses_sale(self) -> None:
src = _quote(("空运费", "1000", "CNY"), total="1000")
ext = extract_adjust("空运费提升到 1100 元", fee_names=["空运费"])
got = apply_adjust(src, ext, business_line="AIR", tms_cost="800", tms_sale="1000")
self.assertAlmostEqual(got.profit_point or 0, 10.0)
def test_profit_uses_manual_not_tms_when_manual_exists(self) -> None:
"""有手工报价:利润点对手工价,不对 TMS。"""
src = _quote(("运输费", "800", "CNY"), total="800")
ext = extract_adjust("运输费提升到 1040 元", fee_names=["运输费"])
got = apply_adjust(src, ext, business_line="SEA", tms_cost="1000", tms_sale="1100")
self.assertAlmostEqual(got.profit_point or 0, 30.0)
def test_profit_air_manual_uses_manual(self) -> None:
src = _quote(("空运费", "2000", "CNY"), total="2000")
ext = extract_adjust("空运费提升到 2200 元", fee_names=["空运费"])
got = apply_adjust(src, ext, business_line="AIR", tms_cost="800", tms_sale="1000")
self.assertAlmostEqual(got.profit_point or 0, 10.0)
def test_profit_second_adjust_keeps_first_baseline(self) -> None:
src = _quote(("运输费", "1000", "CNY"), total="1000")
first = apply_adjust(
src,
extract_adjust("运输费提升到 1100 元", fee_names=["运输费"]),
business_line="LAND",
tms_cost="1000",
)
second = apply_adjust(
first.quote,
extract_adjust("运输费提升到 1300 元", fee_names=["运输费"]),
business_line="LAND",
)
self.assertAlmostEqual(second.profit_point or 0, 30.0)
def test_total_label_cost_until_sales_adjust(self) -> None:
self.assertEqual(total_price_label("AIR"), "总销售价")
self.assertEqual(total_price_label("SEA"), "总成本价")
self.assertEqual(total_price_label("LAND"), "总成本价")
self.assertEqual(total_price_label("", transport_mode="空运"), "总销售价")
self.assertEqual(total_price_label("", transport_mode="陆运"), "总成本价")
tms = {"source": "TMS", "source_label": "TMS标准报价"}
self.assertEqual(total_price_label("LAND", quote=tms), "总成本价")
self.assertEqual(total_price_label("SEA", quote={"source": "PRODUCT"}), "总成本价")
self.assertEqual(total_price_label("LAND", quote={"source": "SALES_ADJUST"}), "总销售价")
self.assertEqual(total_price_label("AIR", quote=tms), "总销售价")
def test_tms_land_card_uses_cost_not_sale(self) -> None:
q = _quote(("运输费", "121.00", "CNY"), total="121.00")
q["source"] = "TMS"
q["source_label"] = "TMS标准报价"
detail = copy.sea_tms_hit_wecom_payload(
work_order_no="WO202609200005", quote=q, business_line="LAND"
)
self.assertEqual(detail["template_card"]["source"]["desc"], "TMS标准报价")
self.assertIn("总成本价:CNY 121.00", detail["template_card"]["quote_area"]["quote_text"])
self.assertNotIn("总销售价", detail["template_card"]["quote_area"]["quote_text"])
scheme = copy.scheme_wecom_payload(
work_order_no="WO202609200005",
facts={"origin": "泉州市", "destination": "南京市"},
quote=q,
transport_mode="陆运",
)
self.assertEqual(scheme["template_card"]["source"]["desc"], "TMS标准报价")
self.assertIn("总成本价:CNY 121.00", scheme["template_card"]["quote_area"]["quote_text"])
deal = copy.deal_horizontal_rows(quote=q, business_line="LAND")
self.assertEqual(deal[0]["keyname"], "总成本价")
def test_adjust_confirm_uses_sale_total_name(self) -> None:
src = _quote(("运输费", "110", "CNY"), total="110")
land = apply_adjust(
src,
extract_adjust("总成本价改为100CNY", fee_names=["运输费"], allow_model=False),
business_line="LAND",
)
self.assertTrue(land.ok)
names = [x.get("name") for x in land.applied]
self.assertIn("总销售价", names)
self.assertNotIn("合计", names)
text = copy.adjust_confirm(land.applied)
self.assertIn("总销售价在当前报价基础上", text)
self.assertNotIn("合计在当前报价基础上", text)
body = copy.adjust_group_body(land.quote, business_line="LAND")
self.assertIn("总销售价:", body)
self.assertNotIn("合计:", body)
air = apply_adjust(
src,
extract_adjust("总销售价改为100CNY", fee_names=["运输费"], allow_model=False),
business_line="AIR",
)
self.assertIn("总销售价", [x.get("name") for x in air.applied])
self.assertNotIn("合计", [x.get("name") for x in air.applied])
note = copy.adjust_ledger_remark(land.applied)
self.assertTrue(note.startswith("销售调价:"))
self.assertIn("总销售价", note)
def test_quote_copy_uses_line_and_source_total_name(self) -> None:
sea = {
"total": "2287.65",
"currency": "CNY",
"source": "TMS",
"fee_rows": [{"name": "海运费", "amount": "100", "currency": "USD"}],
}
air = {
"total": "722.5",
"currency": "CNY",
"source": "TMS",
"fee_rows": [{"name": "空运费", "amount": "700", "currency": "CNY"}],
}
self.assertIn("总成本价:CNY 2287.65", copy.quote_area_fee_text(sea, business_line="SEA"))
self.assertIn("总销售价:CNY 722.5", copy.quote_area_fee_text(air, business_line="AIR"))
self.assertNotIn("总销售价合计", copy.quote_area_fee_text(air, business_line="AIR"))
self.assertNotIn("总销售价:CNY 2287.65", copy.quote_area_fee_text(sea, business_line="SEA"))
self.assertIn("总成本价:CNY 2287.65", "\n".join(copy.quote_detail_lines(sea, business_line="SEA")))
rows = copy.deal_horizontal_rows(quote=air, business_line="AIR")
self.assertEqual(rows[0]["keyname"], "总销售价")
land_rows = copy.deal_horizontal_rows(quote=sea, business_line="LAND")
self.assertEqual(land_rows[0]["keyname"], "总成本价")
def test_adjust_cards_use_sales_adjust_not_tms(self) -> None:
"""调价后价格明细卡、方案卡必须写销售调价,禁止再套 TMS 标准报价。"""
q = _quote(("运输费", "121.00", "CNY"), total="121.00")
q["source"] = "SALES_ADJUST"
q["source_label"] = "销售调价"
detail = copy.sea_tms_hit_wecom_payload(
work_order_no="WO202609200004", quote=q, business_line="LAND"
)
self.assertEqual(detail["template_card"]["source"]["desc"], "销售调价")
self.assertIn("总销售价:CNY 121.00", detail["template_card"]["quote_area"]["quote_text"])
self.assertNotIn("总成本价", detail["template_card"]["quote_area"]["quote_text"])
scheme = copy.scheme_wecom_payload(
work_order_no="WO202609200004",
facts={"origin": "泉州市", "destination": "南京市"},
quote=q,
transport_mode="陆运",
)
self.assertEqual(scheme["template_card"]["source"]["desc"], "销售调价")
self.assertIn("总销售价:CNY 121.00", scheme["template_card"]["quote_area"]["quote_text"])
def test_private_eligibility(self) -> None:
self.assertEqual(private_session_can_adjust("已报价", True, ""), "")
self.assertEqual(private_session_can_adjust("协商中", True, ""), "")
self.assertEqual(private_session_can_adjust("已报价", True, "chat1"), "already_collab")
self.assertEqual(private_session_can_adjust("已成交", True, ""), "closed")
self.assertEqual(private_session_can_adjust("询价中", False, ""), "no_quote")
def test_oral_aliases_not_hardcoded_phrase(self) -> None:
src = _quote(("提货费", "1000", "CNY"), total="1000")
ext = extract_adjust("提货费在现有报价基础上上调 8%", fee_names=["提货费"])
self.assertTrue(ext.is_adjust)
got = apply_adjust(src, ext, business_line="SEA")
self.assertEqual(got.quote["fee_rows"][0]["amount"], "1080.00")
def test_model_phrasing_not_in_local_fallback(self) -> None:
"""口令不写死:模型抽「往上抬一成」也能算;没注入时本地不得靠固定词硬认。"""
spoken = "把运费往上抬一成"
local = extract_adjust(spoken, fee_names=["运输费"], allow_model=False)
self.assertFalse(local.is_adjust)
ext = extract_adjust(
spoken,
fee_names=["运输费"],
injected={
"is_adjust": True,
"items": [{"fee_name": "运输费", "kind": "percent", "direction": "up", "value": 10}],
},
)
got = apply_adjust(_quote(("运输费", "1000", "CNY"), total="1000"), ext, business_line="LAND")
self.assertTrue(got.ok)
self.assertEqual(got.quote["fee_rows"][0]["amount"], "1100.00")
def test_a_route_contract_accepts_quote_adjust(self) -> None:
from agent.routing.deepseek_a import DeepSeekARouter
from agent.routing.ordered_actions import ACTION_INTENTS, parse_ordered_actions_arguments
self.assertEqual(ACTION_INTENTS.get("QUOTE_ADJUST"), "quote_adjust")
parsed = parse_ordered_actions_arguments(
{
"decision_id": "d-adj",
"intent": "quote_adjust",
"ordered_action": {
"action_type": "QUOTE_ADJUST",
"order": 1,
"target": {"work_order_no": "WO202609200002"},
"evidence": {"source_ref": "inbound:u1", "content_digest": "a" * 64},
},
}
)
self.assertEqual(parsed.to_route_decision().intent, "quote_adjust")
router = DeepSeekARouter.from_settings()
sys_msg, user_msg = router.build_messages(
text="报价上升10%",
sender_id="u1",
session_quoted=True,
fee_names=["运输费"],
work_order_no="WO202609200002",
)
self.assertIn("quote_adjust", sys_msg["content"])
self.assertIn("口令不写死", sys_msg["content"])
self.assertIn("session_quoted=true", user_msg["content"])
self.assertIn("运输费", user_msg["content"])
self.assertIn("WO202609200002", user_msg["content"])
def test_private_adjust_stamps_sales_adjust_on_session_quote(self) -> None:
"""书签和出卡必须带销售调价;不能只改主账、卡片仍用 TMS 来源。"""
from agent.policy.quote_adjust_ops import try_private_adjust
class _Sess:
status = "已报价"
phase = "wait_collab"
work_order_no = "WO202609200004"
business_line = "LAND"
collab_chat_id = ""
wait_version = 1
first_or_same = "first"
history_work_order_no = ""
quote = _quote(("运输费", "121.00", "CNY"), total="121.00")
tms_quote = {}
class _Flow:
def __init__(self) -> None:
self.emitted = None
self.saved = None
class _Ledger:
def upsert_product_quote(self, **kwargs):
_Flow.last = kwargs
return {"ok": True}
self.ledger = _Ledger()
def _save(self, sess) -> None:
self.saved = dict(sess.quote or {})
def _emit_sea_hit(self, reply, **kwargs) -> None:
self.emitted = kwargs
_Sess.quote["source"] = "TMS"
_Sess.quote["source_label"] = "TMS标准报价"
sess = _Sess()
flow = _Flow()
replies: list[str] = []
out = try_private_adjust(
flow,
sess,
"运输费提升到 150",
replies.append,
injected_adjust={
"is_adjust": True,
"items": [{"fee_name": "运输费", "kind": "target", "direction": "up", "value": 150}],
},
)
self.assertEqual(out, "wait_collab")
self.assertEqual(sess.quote.get("source_label"), "销售调价")
self.assertEqual(sess.quote.get("source"), "SALES_ADJUST")
self.assertEqual(flow.saved.get("source_label"), "销售调价")
self.assertEqual(flow.last["quote"].get("source_label"), "销售调价")
self.assertEqual(flow.last.get("kind"), "adjust")
self.assertEqual(flow.emitted["quote"].get("source_label"), "销售调价")
self.assertNotIn("TMS", str(flow.emitted["quote"].get("source_label") or ""))
self.assertEqual(str(sess.tms_quote.get("total") or ""), "121.00")
self.assertTrue(copy.quote_is_tms(sess.tms_quote))
self.assertFalse(copy.quote_is_tms(sess.quote))
def test_group_brief_keeps_tms_when_current_is_adjust(self) -> None:
"""拉群摘要:TMS 段必须是原价,调后价另写销售调价,禁止替换。"""
tms = _quote(("运输费", "110.00", "CNY"), total="110.00")
tms["source"] = "TMS"
tms["source_label"] = "TMS标准报价"
adj = _quote(("运输费", "100.00", "CNY"), total="100.00")
adj["source"] = "SALES_ADJUST"
adj["source_label"] = "销售调价"
text = copy.sea_group_brief(
work_order_no="WO202609200006",
facts={"起运地": "泉州市", "目的地": "南京市", "品名": "普货"},
quote=adj,
collab_facts={},
sales_name="销售",
product_names=["产品"],
has_price=True,
tms_quote=tms,
transport_mode="陆运",
)
tms_block = text.split("TMS报价:", 1)[1]
self.assertIn("总成本价:CNY 110.00", tms_block.split("销售调价:", 1)[0])
self.assertIn("销售调价:", text)
self.assertIn("总销售价:CNY 100.00", text.split("销售调价:", 1)[1])
self.assertNotIn("总销售价:CNY 100.00", tms_block.split("销售调价:", 1)[0])
def test_group_brief_adjust_alone_does_not_become_tms(self) -> None:
"""没单独传入 TMS 时,调后价也不能写进 TMS 段。"""
adj = _quote(("运输费", "100.00", "CNY"), total="100.00")
adj["source"] = "SALES_ADJUST"
adj["source_label"] = "销售调价"
text = copy.sea_group_brief(
work_order_no="WO202609200006",
facts={"起运地": "泉州市", "目的地": "南京市"},
quote=adj,
collab_facts={},
sales_name="销售",
product_names=[],
has_price=True,
transport_mode="陆运",
)
self.assertIn("TMS报价:\nTMS 暂无匹配报价", text)
self.assertIn("销售调价:", text)
self.assertIn("总销售价:CNY 100.00", text.split("销售调价:", 1)[1])
if __name__ == "__main__":
unittest.main()