547 lines
20 KiB
Python
547 lines
20 KiB
Python
"""
|
|
陆运线路：按 segmentSequence 成条；一段两段是一条带中转，不是两条。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
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.land_route_options import (
|
|
group_land_routes,
|
|
needs_route_pick,
|
|
option_fee_lines,
|
|
option_total,
|
|
project_selected_route,
|
|
route_line,
|
|
)
|
|
|
|
|
|
def _single_with_transfer() -> dict:
|
|
return {
|
|
"isMulti": 1,
|
|
"landSegments": [
|
|
{
|
|
"segmentSequence": 1,
|
|
"routeLegSequence": 1,
|
|
"origin": "上海",
|
|
"destination": "谅山",
|
|
"costPrice": 7500,
|
|
"currency": "CNY",
|
|
"transitDays": 48,
|
|
},
|
|
{
|
|
"segmentSequence": 1,
|
|
"routeLegSequence": 2,
|
|
"origin": "谅山",
|
|
"destination": "胡志明",
|
|
"costPrice": 9600,
|
|
"currency": "CNY",
|
|
"transitDays": 72,
|
|
},
|
|
],
|
|
}
|
|
|
|
|
|
def _three_routes() -> dict:
|
|
base = _single_with_transfer()
|
|
base["isMulti"] = 2
|
|
extra = [
|
|
{
|
|
"segmentSequence": 2,
|
|
"routeLegSequence": 1,
|
|
"origin": "上海",
|
|
"destination": "深圳",
|
|
"costPrice": 7500,
|
|
"currency": "CNY",
|
|
"transitDays": 48,
|
|
},
|
|
{
|
|
"segmentSequence": 2,
|
|
"routeLegSequence": 2,
|
|
"origin": "深圳",
|
|
"destination": "胡志明",
|
|
"costPrice": 9600,
|
|
"currency": "CNY",
|
|
"transitDays": 72,
|
|
},
|
|
{
|
|
"segmentSequence": 3,
|
|
"routeLegSequence": 1,
|
|
"origin": "上海",
|
|
"destination": "广西",
|
|
"costPrice": 7500,
|
|
"currency": "CNY",
|
|
"transitDays": 48,
|
|
},
|
|
{
|
|
"segmentSequence": 3,
|
|
"routeLegSequence": 2,
|
|
"origin": "广西",
|
|
"destination": "胡志明",
|
|
"costPrice": 9600,
|
|
"currency": "CNY",
|
|
"transitDays": 72,
|
|
},
|
|
]
|
|
base["landSegments"] = list(base["landSegments"]) + extra
|
|
return base
|
|
|
|
|
|
def _tms_land_fees(*pairs: tuple[int, int, str, int]) -> list[dict]:
|
|
"""TMS feeItems 原文：费用名按回包，不编路段名。"""
|
|
rows: list[dict] = []
|
|
for seq, leg, name, amount in pairs:
|
|
rows.append(
|
|
{
|
|
"feeName": name,
|
|
"name": name,
|
|
"amount": amount,
|
|
"currency": "CNY",
|
|
"segmentSequence": seq,
|
|
"routeLegSequence": leg,
|
|
"unit": "台",
|
|
"quantity": 1,
|
|
}
|
|
)
|
|
return rows
|
|
|
|
|
|
class LandRouteOptionsTests(unittest.TestCase):
|
|
def test_two_legs_one_route(self) -> None:
|
|
quote = _single_with_transfer()
|
|
routes = group_land_routes(quote)
|
|
self.assertEqual(len(routes), 1)
|
|
self.assertFalse(needs_route_pick(quote))
|
|
self.assertEqual(route_line(routes[0]), "上海-谅山-胡志明")
|
|
fees = option_fee_lines(routes[0])
|
|
self.assertEqual(fees[0][0], "上海 → 谅山")
|
|
self.assertIn("7500", fees[0][1])
|
|
self.assertEqual(fees[1][0], "谅山 → 胡志明")
|
|
self.assertIn("17100", option_total(routes[0]))
|
|
|
|
def test_three_segment_sequences_need_h5(self) -> None:
|
|
quote = _three_routes()
|
|
routes = group_land_routes(quote)
|
|
self.assertEqual(len(routes), 3)
|
|
self.assertTrue(needs_route_pick(quote))
|
|
self.assertEqual(route_line(routes[1]), "上海-深圳-胡志明")
|
|
|
|
def test_project_writes_n_plus_eta_not_hours(self) -> None:
|
|
quote = {
|
|
"isMulti": 1,
|
|
"eta": "-",
|
|
"landSegments": [
|
|
{
|
|
"segmentSequence": 1,
|
|
"routeLegSequence": 1,
|
|
"origin": "上海",
|
|
"destination": "谅山",
|
|
"costPrice": 7500,
|
|
"currency": "CNY",
|
|
"transitDays": "N+2",
|
|
},
|
|
{
|
|
"segmentSequence": 1,
|
|
"routeLegSequence": 2,
|
|
"origin": "谅山",
|
|
"destination": "胡志明",
|
|
"costPrice": 9600,
|
|
"currency": "CNY",
|
|
"transitDays": "N+3",
|
|
},
|
|
],
|
|
}
|
|
routes = group_land_routes(quote)
|
|
projected = project_selected_route(quote, routes[0])
|
|
eta = str(projected.get("eta") or "")
|
|
self.assertIn("N+2", eta)
|
|
self.assertIn("N+3", eta)
|
|
self.assertIn("17100", str(projected.get("total") or ""))
|
|
|
|
def test_project_keeps_tms_fee_names_not_od(self) -> None:
|
|
quote = _single_with_transfer()
|
|
quote["fee_lines"] = _tms_land_fees(
|
|
(1, 1, "运输费", 7500),
|
|
(1, 2, "运输费", 9600),
|
|
)
|
|
routes = group_land_routes(quote)
|
|
projected = project_selected_route(quote, routes[0])
|
|
names = [x.get("name") for x in projected.get("fee_lines") or []]
|
|
self.assertEqual(names, ["运输费", "运输费"])
|
|
joined = json.dumps(projected.get("fee_lines"), ensure_ascii=False)
|
|
self.assertNotIn("上海 →", joined)
|
|
self.assertIn("7500", joined)
|
|
self.assertIn("9600", joined)
|
|
|
|
def test_project_keeps_only_selected_legs(self) -> None:
|
|
quote = _three_routes()
|
|
quote["fee_lines"] = _tms_land_fees(
|
|
(1, 1, "运输费", 7500),
|
|
(1, 2, "运输费", 9600),
|
|
(2, 1, "运输费", 7500),
|
|
(2, 2, "运输费", 9600),
|
|
(3, 1, "提货费", 500),
|
|
(3, 2, "运输费", 9600),
|
|
)
|
|
routes = group_land_routes(quote)
|
|
projected = project_selected_route(quote, routes[1])
|
|
raw = projected.get("landSegments")
|
|
if isinstance(raw, str):
|
|
raw = json.loads(raw)
|
|
seqs = {int(x.get("segmentSequence") or 0) for x in raw}
|
|
self.assertEqual(seqs, {2})
|
|
names = [x.get("name") for x in projected.get("fee_lines") or []]
|
|
self.assertEqual(names, ["运输费", "运输费"])
|
|
self.assertIn("17100", str(projected.get("total") or ""))
|
|
self.assertNotIn("谅山", json.dumps(projected.get("fee_lines"), ensure_ascii=False))
|
|
self.assertNotIn("提货费", json.dumps(projected.get("fee_lines"), ensure_ascii=False))
|
|
self.assertNotIn("上海 →", json.dumps(projected.get("fee_lines"), ensure_ascii=False))
|
|
|
|
def test_detail_puts_tms_fee_under_each_leg(self) -> None:
|
|
from agent.policy import inquiry_copy as copy
|
|
|
|
quote = _single_with_transfer()
|
|
quote["landSegments"][0]["origin"] = "上海"
|
|
quote["landSegments"][0]["destination"] = "谅山Lạng Sơn-春疆货场Huu Nghi"
|
|
quote["landSegments"][0]["transitDays"] = "N+2"
|
|
quote["landSegments"][1]["origin"] = "谅山Lạng Sơn-春疆货场Huu Nghi"
|
|
quote["landSegments"][1]["destination"] = "胡志明Hồ Chí Minh"
|
|
quote["landSegments"][1]["transitDays"] = "N+3"
|
|
quote["fee_lines"] = _tms_land_fees(
|
|
(1, 1, "运输费", 7500.0),
|
|
(1, 2, "运输费", 9600.0),
|
|
)
|
|
routes = group_land_routes(quote)
|
|
projected = project_selected_route(quote, routes[0])
|
|
text = copy.sea_quote_detail_text(
|
|
work_order_no="WO202609220029",
|
|
quote=projected,
|
|
business_line="LAND",
|
|
)
|
|
self.assertIn("上海-谅山Lạng Sơn-春疆货场Huu Nghi-胡志明Hồ Chí Minh", text)
|
|
od1 = text.find("上海 → 谅山Lạng Sơn-春疆货场Huu Nghi")
|
|
fee1 = text.find("运输费：CNY 7500", od1)
|
|
od2 = text.find("谅山Lạng Sơn-春疆货场Huu Nghi → 胡志明Hồ Chí Minh")
|
|
fee2 = text.find("运输费：CNY 9600", od2)
|
|
self.assertGreaterEqual(od1, 0)
|
|
self.assertGreater(fee1, od1)
|
|
self.assertGreater(od2, fee1)
|
|
self.assertGreater(fee2, od2)
|
|
eta1 = text.find("时效：N+2", fee1)
|
|
eta2 = text.find("时效：N+3", fee2)
|
|
self.assertGreater(eta1, fee1)
|
|
self.assertLess(eta1, od2)
|
|
self.assertGreater(eta2, fee2)
|
|
self.assertIn("总成本价：CNY 17100", text)
|
|
|
|
def test_h5_lists_segment_costs(self) -> None:
|
|
from agent.channel.h5.land_route_page import render_land_route_page
|
|
|
|
html = render_land_route_page(
|
|
{"work_order_no": "WO202609220010", "quote": _three_routes()}
|
|
)
|
|
self.assertIn("上海-深圳-胡志明", html)
|
|
self.assertIn("上海 → 深圳", html)
|
|
self.assertIn("成本价合计", html)
|
|
self.assertNotIn("userid", html.lower())
|
|
|
|
def test_h5_puts_tms_fee_under_each_leg(self) -> None:
|
|
from agent.channel.h5.land_route_page import render_land_route_page
|
|
|
|
quote = _three_routes()
|
|
for leg in quote["landSegments"]:
|
|
if str(leg.get("destination") or "") == "深圳":
|
|
leg["transitDays"] = "N+2"
|
|
if str(leg.get("origin") or "") == "深圳":
|
|
leg["transitDays"] = "N+3"
|
|
quote["fee_lines"] = _tms_land_fees(
|
|
(1, 1, "运输费", 7500),
|
|
(1, 2, "运输费", 9600),
|
|
(2, 1, "运输费", 7500),
|
|
(2, 2, "运输费", 9600),
|
|
(3, 1, "运输费", 7500),
|
|
(3, 2, "运输费", 9600),
|
|
)
|
|
page = render_land_route_page(
|
|
{"work_order_no": "WO202609220010", "quote": quote}
|
|
)
|
|
od = page.find("上海 → 深圳")
|
|
fee = page.find("运输费", od)
|
|
self.assertGreaterEqual(od, 0)
|
|
self.assertGreater(fee, od)
|
|
eta = page.find("时效", fee)
|
|
self.assertGreater(eta, fee)
|
|
self.assertIn("N+2", page)
|
|
self.assertIn("N+3", page)
|
|
self.assertIn("CNY 7500", page)
|
|
self.assertIn("成本价合计", page)
|
|
|
|
def test_h5_and_detail_use_tms_fees_when_java_sends_json_strings(self) -> None:
|
|
"""主账 quote 常是 JSON 串，且 fee_lines 可能没带段号。"""
|
|
from agent.channel.h5.land_route_page import render_land_route_page
|
|
from agent.policy import inquiry_copy as copy
|
|
|
|
quote = {
|
|
"isMulti": "2",
|
|
"currency": "CNY",
|
|
"landSegments": json.dumps(
|
|
[
|
|
{
|
|
"segmentSequence": 1,
|
|
"routeLegSequence": 1,
|
|
"origin": "上海",
|
|
"destination": "谅山Lạng Sơn-春疆货场Huu Nghi",
|
|
"costPrice": 7500.0,
|
|
"currency": "CNY",
|
|
"transitDays": "N+2",
|
|
},
|
|
{
|
|
"segmentSequence": 1,
|
|
"routeLegSequence": 2,
|
|
"origin": "谅山Lạng Sơn-春疆货场Huu Nghi",
|
|
"destination": "胡志明Hồ Chí Minh",
|
|
"costPrice": 9600.0,
|
|
"currency": "CNY",
|
|
"transitDays": "N+3",
|
|
},
|
|
{
|
|
"segmentSequence": 2,
|
|
"routeLegSequence": 1,
|
|
"origin": "上海",
|
|
"destination": "广西",
|
|
"costPrice": 7500.0,
|
|
"currency": "CNY",
|
|
"transitDays": "N+3",
|
|
},
|
|
{
|
|
"segmentSequence": 2,
|
|
"routeLegSequence": 2,
|
|
"origin": "广西",
|
|
"destination": "胡志明Hồ Chí Minh",
|
|
"costPrice": 12500.0,
|
|
"currency": "CNY",
|
|
"transitDays": "N+3",
|
|
},
|
|
],
|
|
ensure_ascii=False,
|
|
),
|
|
"fee_lines": json.dumps(
|
|
[
|
|
{"name": "运输费", "amount": "7500.0", "currency": "CNY"},
|
|
{"name": "运输费", "amount": "9600.0", "currency": "CNY"},
|
|
{"name": "运输费", "amount": "7500.0", "currency": "CNY"},
|
|
{"name": "运输费", "amount": "12500.0", "currency": "CNY"},
|
|
],
|
|
ensure_ascii=False,
|
|
),
|
|
}
|
|
page = render_land_route_page(
|
|
{"work_order_no": "WO202609220036", "quote": quote}
|
|
)
|
|
od = page.find("上海 → 谅山Lạng Sơn-春疆货场Huu Nghi")
|
|
fee = page.find("运输费", od)
|
|
eta = page.find("时效", fee)
|
|
self.assertGreaterEqual(od, 0)
|
|
self.assertGreater(fee, od)
|
|
self.assertGreater(eta, fee)
|
|
self.assertIn("N+2", page)
|
|
routes = group_land_routes(quote)
|
|
projected = project_selected_route(quote, routes[0])
|
|
text = copy.sea_quote_detail_text(
|
|
work_order_no="WO202609220036",
|
|
quote=projected,
|
|
business_line="LAND",
|
|
)
|
|
self.assertIn("运输费：CNY 7500", text)
|
|
self.assertIn("时效：N+2", text)
|
|
self.assertIn("时效：N+3", text)
|
|
self.assertNotRegex(text, r"上海 → 谅山.+:CNY")
|
|
|
|
def test_recovers_transport_fee_when_fee_lines_are_od_names(self) -> None:
|
|
"""测服 WO036 这类：fee_lines 被写成路段名，销售侧仍要看到运输费和时效。"""
|
|
from agent.channel.h5.land_route_page import render_land_route_page
|
|
from agent.policy import inquiry_copy as copy
|
|
|
|
quote = {
|
|
"isMulti": "2",
|
|
"currency": "CNY",
|
|
"route_line": "上海-谅山Lạng Sơn-春疆货场Huu Nghi-胡志明Hồ Chí Minh",
|
|
"landSegments": [
|
|
{
|
|
"segmentSequence": 1,
|
|
"routeLegSequence": 1,
|
|
"origin": "上海",
|
|
"destination": "谅山Lạng Sơn-春疆货场Huu Nghi",
|
|
"costPrice": 7500.0,
|
|
"currency": "CNY",
|
|
"transitDays": "N+2",
|
|
},
|
|
{
|
|
"segmentSequence": 1,
|
|
"routeLegSequence": 2,
|
|
"origin": "谅山Lạng Sơn-春疆货场Huu Nghi",
|
|
"destination": "胡志明Hồ Chí Minh",
|
|
"costPrice": 9600.0,
|
|
"currency": "CNY",
|
|
"transitDays": "N+3",
|
|
},
|
|
{
|
|
"segmentSequence": 2,
|
|
"routeLegSequence": 1,
|
|
"origin": "上海",
|
|
"destination": "广西",
|
|
"costPrice": 7500.0,
|
|
"currency": "CNY",
|
|
"transitDays": "N+3",
|
|
},
|
|
{
|
|
"segmentSequence": 2,
|
|
"routeLegSequence": 2,
|
|
"origin": "广西",
|
|
"destination": "胡志明Hồ Chí Minh",
|
|
"costPrice": 12500.0,
|
|
"currency": "CNY",
|
|
"transitDays": "N+3",
|
|
},
|
|
],
|
|
"fee_lines": [
|
|
{"name": "上海 → 谅山Lạng Sơn-春疆货场Huu Nghi", "amount": "7500.0", "currency": "CNY"},
|
|
{"name": "谅山Lạng Sơn-春疆货场Huu Nghi → 胡志明Hồ Chí Minh", "amount": "9600.0", "currency": "CNY"},
|
|
],
|
|
}
|
|
page = render_land_route_page(
|
|
{"work_order_no": "WO202609220036", "quote": quote}
|
|
)
|
|
od = page.find("上海 → 谅山Lạng Sơn-春疆货场Huu Nghi")
|
|
fee = page.find("运输费", od)
|
|
eta = page.find("时效", fee)
|
|
self.assertGreaterEqual(od, 0)
|
|
self.assertGreater(fee, od)
|
|
self.assertGreater(eta, fee)
|
|
self.assertIn("N+2", page)
|
|
routes = group_land_routes(quote)
|
|
projected = project_selected_route(quote, routes[0])
|
|
text = copy.sea_quote_detail_text(
|
|
work_order_no="WO202609220036",
|
|
quote=projected,
|
|
business_line="LAND",
|
|
)
|
|
self.assertIn("运输费：CNY 7500", text)
|
|
self.assertIn("时效：N+2", text)
|
|
self.assertIn("时效：N+3", text)
|
|
self.assertNotRegex(text, r"上海 → 谅山.+:CNY")
|
|
|
|
def test_single_emits_detail_multi_opens_h5(self) -> None:
|
|
from agent.ledger.memory_ledger import MemoryLedger
|
|
from agent.policy.land_text_flow import LandTextInquiryFlow
|
|
|
|
def _go(flag: str) -> tuple[str, list[str]]:
|
|
flow = LandTextInquiryFlow(MemoryLedger())
|
|
flow._detail_before_card_gap_sec = 0
|
|
replies: list[str] = []
|
|
|
|
def reply(text: str, extra=None) -> str:
|
|
replies.append(text)
|
|
_ = extra
|
|
return ""
|
|
|
|
facts = {
|
|
"起运港": "上海",
|
|
"目的港": "胡志明",
|
|
"品名": "衣服",
|
|
"体积": "100CBM",
|
|
"毛重": "50kg",
|
|
flag: "1",
|
|
}
|
|
flow.on_text(
|
|
sender_id="u_land_rt",
|
|
text="国内运输拼车 国内长途/零担 急件",
|
|
reply=reply,
|
|
injected_facts=facts,
|
|
injected_mode="LAND",
|
|
)
|
|
phase = flow.on_text(sender_id="u_land_rt", text="确定", reply=reply)
|
|
return phase, replies
|
|
|
|
phase, replies = _go("tms_land_single")
|
|
self.assertEqual(phase, "wait_collab")
|
|
self.assertTrue(any("运输费" in x for x in replies))
|
|
self.assertTrue(any("上海-谅山-胡志明" in x for x in replies))
|
|
self.assertTrue(any("上海 → 谅山" in x for x in replies))
|
|
phase, replies = _go("tms_land_multi")
|
|
self.assertEqual(phase, "wait_land_route")
|
|
self.assertTrue(any("条线路" in x for x in replies))
|
|
|
|
def test_select_then_reselect_after_adopt_voids_old_file(self) -> None:
|
|
from agent.ledger.memory_ledger import MemoryLedger
|
|
from agent.policy import inquiry_copy as copy
|
|
from agent.policy.land_text_flow import LandTextInquiryFlow
|
|
|
|
ledger = MemoryLedger()
|
|
flow = LandTextInquiryFlow(ledger)
|
|
flow._detail_before_card_gap_sec = 0
|
|
replies: list[str] = []
|
|
|
|
def reply(text: str, extra=None) -> str:
|
|
replies.append(text)
|
|
_ = extra
|
|
return ""
|
|
|
|
facts = {
|
|
"起运港": "上海",
|
|
"目的港": "胡志明",
|
|
"品名": "衣服",
|
|
"体积": "100CBM",
|
|
"毛重": "50kg",
|
|
"tms_land_multi": "1",
|
|
}
|
|
flow.on_text(
|
|
sender_id="u_land_void",
|
|
text="国内运输拼车 国内长途/零担 急件",
|
|
reply=reply,
|
|
injected_facts=facts,
|
|
injected_mode="LAND",
|
|
)
|
|
self.assertEqual(
|
|
flow.on_text(sender_id="u_land_void", text="确定", reply=reply),
|
|
"wait_land_route",
|
|
)
|
|
self.assertEqual(
|
|
flow.select_land_route(sender_id="u_land_void", index=1, reply=reply),
|
|
"wait_collab",
|
|
)
|
|
joined = "\n".join(replies)
|
|
self.assertIn("上海-深圳-胡志明", joined)
|
|
self.assertIn("运输费", joined)
|
|
self.assertIn("上海 → 深圳", joined)
|
|
self.assertNotIn("谅山", "\n".join(replies[-3:]))
|
|
self.assertEqual(
|
|
flow.on_button(sender_id="u_land_void", action=copy.BTN_SKIP_COLLAB, reply=reply),
|
|
"wait_adopt",
|
|
)
|
|
self.assertEqual(
|
|
flow.on_button(sender_id="u_land_void", action=copy.BTN_ADOPT_EXCEL, reply=reply),
|
|
"wait_deal",
|
|
)
|
|
wo = flow.session_of("u_land_void").work_order_no
|
|
self.assertTrue(ledger.get_ticket(work_order_no=wo).quote_file_ready)
|
|
phase = flow.select_land_route(sender_id="u_land_void", index=0, reply=reply)
|
|
self.assertEqual(phase, "wait_adopt")
|
|
self.assertFalse(ledger.get_ticket(work_order_no=wo).quote_file_ready)
|
|
ticket = ledger.get_ticket(work_order_no=wo)
|
|
segs = ticket.quote.get("landSegments") or []
|
|
seqs = {int(x.get("segmentSequence") or 0) for x in segs}
|
|
self.assertEqual(seqs, {1})
|
|
self.assertTrue(any("报价方案" in x or "生成Excel" in x for x in replies[-3:]))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|