421 lines
14 KiB
Python
421 lines
14 KiB
Python
"""
|
|
陆运 TMS 线路报价：按 segmentSequence 收成一条条完整线路。
|
|
|
|
本文件职责：一段里的多段是中转，不是多条线路；抽出每段成本；选定后只留这一条的段。
|
|
禁止：把 costPrice 标成销售价；编没有回的站名；改六态。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _as_list(raw: Any) -> list[Any]:
|
|
"""list 或 JSON 串。"""
|
|
if isinstance(raw, str) and raw.strip():
|
|
try:
|
|
raw = json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
logger.warning("land.routes landSegments 不是 JSON")
|
|
return []
|
|
return list(raw) if isinstance(raw, list) else []
|
|
|
|
|
|
def parse_land_legs(quote: dict[str, Any] | None) -> list[dict[str, Any]]:
|
|
"""摊平 quote.landSegments。"""
|
|
src = dict(quote or {})
|
|
out: list[dict[str, Any]] = []
|
|
for item in _as_list(src.get("landSegments") or src.get("land_segments")):
|
|
if isinstance(item, dict):
|
|
out.append(dict(item))
|
|
return out
|
|
|
|
|
|
def _seq(item: dict[str, Any]) -> int:
|
|
try:
|
|
return int(item.get("segmentSequence") or item.get("segment_sequence") or 1)
|
|
except (TypeError, ValueError):
|
|
return 1
|
|
|
|
|
|
def _leg_no(item: dict[str, Any]) -> int:
|
|
try:
|
|
return int(item.get("routeLegSequence") or item.get("route_leg_sequence") or 0)
|
|
except (TypeError, ValueError):
|
|
return 0
|
|
|
|
|
|
def group_land_routes(quote: dict[str, Any] | None) -> list[dict[str, Any]]:
|
|
"""
|
|
同一 segmentSequence 收成一条线路，段按 routeLegSequence 排。
|
|
"""
|
|
buckets: dict[int, list[dict[str, Any]]] = {}
|
|
order: list[int] = []
|
|
for leg in parse_land_legs(quote):
|
|
key = _seq(leg)
|
|
if key not in buckets:
|
|
buckets[key] = []
|
|
order.append(key)
|
|
buckets[key].append(leg)
|
|
routes: list[dict[str, Any]] = []
|
|
for key in order:
|
|
legs = sorted(buckets[key], key=_leg_no)
|
|
routes.append({"segmentSequence": key, "legs": legs})
|
|
return routes
|
|
|
|
|
|
def needs_route_pick(quote: dict[str, Any] | None) -> bool:
|
|
"""多于一条完整线路才进 H5。"""
|
|
return len(group_land_routes(quote)) > 1
|
|
|
|
|
|
def route_line(route: dict[str, Any] | None) -> str:
|
|
"""上海-谅山-胡志明：第一段起运 + 各段目的。"""
|
|
legs = list((route or {}).get("legs") or [])
|
|
if not legs:
|
|
return ""
|
|
names = [str(legs[0].get("origin") or "").strip()]
|
|
for leg in legs:
|
|
dest = str(leg.get("destination") or "").strip()
|
|
if dest:
|
|
names.append(dest)
|
|
return "-".join(x for x in names if x)
|
|
|
|
|
|
def _money(raw: Any, currency: str = "") -> str:
|
|
text = str(raw if raw is not None else "").strip()
|
|
if not text or text in {"None", "null"}:
|
|
return ""
|
|
ccy = (currency or "").strip()
|
|
if ccy and ccy not in text:
|
|
return f"{ccy} {text}"
|
|
return text
|
|
|
|
|
|
def option_fee_lines(route: dict[str, Any] | None) -> list[tuple[str, str]]:
|
|
"""选线路页用：始发 → 目的 + 该段成本，方便区分备选。报价明细不走这里。"""
|
|
rows: list[tuple[str, str]] = []
|
|
for leg in list((route or {}).get("legs") or []):
|
|
origin = str(leg.get("origin") or "").strip() or "-"
|
|
dest = str(leg.get("destination") or "").strip() or "-"
|
|
ccy = str(leg.get("currency") or "").strip()
|
|
amount = _money(leg.get("costPrice") if leg.get("costPrice") is not None else leg.get("cost_price"), ccy)
|
|
if not amount:
|
|
continue
|
|
rows.append((f"{origin} → {dest}", amount))
|
|
return rows
|
|
|
|
|
|
def _num(raw: Any) -> float | None:
|
|
"""金额比大小：去掉币种和逗号。"""
|
|
if raw is None:
|
|
return None
|
|
text = str(raw).strip()
|
|
if not text or text in {"-", "None", "null"}:
|
|
return None
|
|
for bit in text.replace(",", " ").split():
|
|
try:
|
|
return float(bit)
|
|
except ValueError:
|
|
continue
|
|
return None
|
|
|
|
|
|
def _same_amount(left: Any, right: Any) -> bool:
|
|
a, b = _num(left), _num(right)
|
|
if a is None or b is None:
|
|
return False
|
|
return abs(a - b) < 0.0001
|
|
|
|
|
|
def _leg_cost(leg: dict[str, Any] | None) -> Any:
|
|
src = dict(leg or {})
|
|
if src.get("costPrice") is not None:
|
|
return src.get("costPrice")
|
|
return src.get("cost_price")
|
|
|
|
|
|
def _has_seq(item: dict[str, Any] | None) -> bool:
|
|
src = dict(item or {})
|
|
raw = src.get("segmentSequence")
|
|
if raw is None:
|
|
raw = src.get("segment_sequence")
|
|
return raw not in (None, "", "null")
|
|
|
|
|
|
def _fee_name(item: dict[str, Any] | None) -> str:
|
|
"""费用名只认 TMS 原文，禁止用路段名冒充。"""
|
|
src = dict(item or {})
|
|
return str(
|
|
src.get("feeName")
|
|
or src.get("name")
|
|
or src.get("chargeName")
|
|
or src.get("itemName")
|
|
or src.get("feeType")
|
|
or ""
|
|
).strip()
|
|
|
|
|
|
def parse_tms_fee_rows(quote: dict[str, Any] | None) -> list[dict[str, Any]]:
|
|
"""摊平 TMS feeItems / 已投影的 fee_lines。"""
|
|
src = dict(quote or {})
|
|
raw = src.get("feeItems") or src.get("fee_items") or src.get("fee_lines") or src.get("fee_rows")
|
|
return [dict(x) for x in _as_list(raw) if isinstance(x, dict)]
|
|
|
|
|
|
def _named_from_rows(raw: Any) -> list[dict[str, Any]]:
|
|
"""只要 TMS 费目名。路段名冒充的费用行丢掉。"""
|
|
rows: list[dict[str, Any]] = []
|
|
for item in _as_list(raw):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
name = _fee_name(item)
|
|
if not name or "→" in name or "->" in name:
|
|
continue
|
|
amount = item.get("amount")
|
|
if amount is None:
|
|
amount = item.get("costPrice") if item.get("costPrice") is not None else item.get("salePrice")
|
|
if _num(amount) is None:
|
|
continue
|
|
line = dict(item)
|
|
line["name"] = name
|
|
line["amount"] = amount
|
|
rows.append(line)
|
|
return rows
|
|
|
|
|
|
def _named_tms_fees(quote: dict[str, Any] | None) -> list[dict[str, Any]]:
|
|
"""
|
|
先看已投影的 fee_lines，没有真费目再看 feeItems。
|
|
|
|
主账有时把路段名写进 fee_lines；这时必须回到 TMS feeItems 取运输费。
|
|
"""
|
|
src = dict(quote or {})
|
|
named = _named_from_rows(src.get("fee_lines") or src.get("fee_rows"))
|
|
if named:
|
|
return named
|
|
named = _named_from_rows(src.get("feeItems") or src.get("fee_items"))
|
|
if named:
|
|
return named
|
|
return _named_from_rows(parse_tms_fee_rows(src))
|
|
|
|
|
|
def _pick_fee_for_leg(
|
|
fees: list[dict[str, Any]],
|
|
used: set[int],
|
|
leg: dict[str, Any],
|
|
prefer_seq: int = 0,
|
|
) -> dict[str, Any] | None:
|
|
"""按段号 / 腿号 / 金额认回这一段的 TMS 费用。"""
|
|
want_leg = _leg_no(leg)
|
|
want_amt = _leg_cost(leg)
|
|
checks = (
|
|
lambda f: prefer_seq and _has_seq(f) and _seq(f) == prefer_seq and _leg_no(f) == want_leg,
|
|
lambda f: prefer_seq and _has_seq(f) and _seq(f) == prefer_seq and _same_amount(f.get("amount"), want_amt),
|
|
lambda f: _leg_no(f) == want_leg and _same_amount(f.get("amount"), want_amt),
|
|
lambda f: _same_amount(f.get("amount"), want_amt),
|
|
)
|
|
for check in checks:
|
|
for i, fee in enumerate(fees):
|
|
if i in used:
|
|
continue
|
|
try:
|
|
ok = check(fee)
|
|
except (TypeError, ValueError):
|
|
ok = False
|
|
if ok:
|
|
used.add(i)
|
|
return fee
|
|
return None
|
|
|
|
|
|
def selected_tms_fee_lines(
|
|
quote: dict[str, Any] | None,
|
|
route: dict[str, Any] | None,
|
|
) -> list[dict[str, Any]]:
|
|
"""
|
|
选定一条陆运线路后：只留该条 TMS 费用行，名称原样。
|
|
|
|
主账常把 fee_lines 收成 JSON 串，也不一定带段号；这时按各段成本和腿号认回去。
|
|
"""
|
|
src = dict(route or {})
|
|
try:
|
|
seq = int(src.get("segmentSequence") or src.get("segment_sequence") or 0)
|
|
except (TypeError, ValueError):
|
|
seq = 0
|
|
fees = _named_tms_fees(quote)
|
|
matched = [f for f in fees if _has_seq(f) and _seq(f) == seq]
|
|
if matched:
|
|
return matched
|
|
used: set[int] = set()
|
|
out: list[dict[str, Any]] = []
|
|
for leg in list(src.get("legs") or []):
|
|
if not isinstance(leg, dict):
|
|
continue
|
|
picked = _pick_fee_for_leg(fees, used, leg, prefer_seq=seq)
|
|
if picked:
|
|
out.append(picked)
|
|
if out:
|
|
return out
|
|
if fees and not needs_route_pick(quote):
|
|
return fees
|
|
return []
|
|
|
|
|
|
def _od_title(leg: dict[str, Any] | None) -> str:
|
|
"""上海 → 谅山：站名用 TMS 原文。"""
|
|
src = dict(leg or {})
|
|
origin = str(src.get("origin") or "").strip()
|
|
dest = str(src.get("destination") or "").strip()
|
|
if not origin and not dest:
|
|
return ""
|
|
return f"{origin or '-'} → {dest or '-'}"
|
|
|
|
|
|
def _quote_fee_dicts(quote: dict[str, Any] | None) -> list[dict[str, Any]]:
|
|
"""优先已选定的 fee_lines，避免再摊开整单 feeItems。"""
|
|
src = dict(quote or {})
|
|
rows = [dict(x) for x in _as_list(src.get("fee_lines") or src.get("fee_rows")) if isinstance(x, dict)]
|
|
if rows:
|
|
return rows
|
|
return parse_tms_fee_rows(src)
|
|
|
|
|
|
def _selected_legs(quote: dict[str, Any] | None) -> list[dict[str, Any]]:
|
|
"""当前这条线路的各段。多条未选时不编。"""
|
|
src = dict(quote or {})
|
|
selected = src.get("selected_land_route")
|
|
if isinstance(selected, dict):
|
|
legs = [dict(x) for x in list(selected.get("legs") or []) if isinstance(x, dict)]
|
|
if legs:
|
|
return legs
|
|
routes = group_land_routes(src)
|
|
if len(routes) == 1:
|
|
return [dict(x) for x in list(routes[0].get("legs") or []) if isinstance(x, dict)]
|
|
legs = parse_land_legs(src)
|
|
seqs = {_seq(x) for x in legs}
|
|
if len(seqs) == 1:
|
|
return sorted(legs, key=_leg_no)
|
|
return []
|
|
|
|
|
|
def land_leg_fee_blocks(
|
|
quote: dict[str, Any] | None,
|
|
) -> list[tuple[str, list[dict[str, Any]], str]]:
|
|
"""
|
|
价格明细按段：路段标题、该段 TMS 费目、该段时效原文。
|
|
|
|
例：上海 → 谅山 / 运输费 7500 / N+2。时效没有就不写，禁止编。
|
|
"""
|
|
legs = _selected_legs(quote)
|
|
if not legs:
|
|
return []
|
|
fees = _named_tms_fees(quote)
|
|
fallback_name = next((str(f.get("name") or "").strip() for f in fees if f.get("name")), "")
|
|
# 路段名把 TMS 费目冲掉时，各段成本仍按运输费展示，禁止再用始发→目的冒充费名。
|
|
if not fallback_name:
|
|
fallback_name = "运输费"
|
|
used: set[int] = set()
|
|
blocks: list[tuple[str, list[dict[str, Any]], str]] = []
|
|
for leg in legs:
|
|
title = _od_title(leg)
|
|
if not title:
|
|
continue
|
|
eta = str(leg.get("transitDays") or leg.get("transit_days") or "").strip()
|
|
picked: list[dict[str, Any]] = []
|
|
one = _pick_fee_for_leg(fees, used, leg, prefer_seq=_seq(leg))
|
|
if one:
|
|
picked.append(one)
|
|
elif fallback_name and _num(_leg_cost(leg)) is not None:
|
|
picked.append(
|
|
{
|
|
"name": fallback_name,
|
|
"amount": _leg_cost(leg),
|
|
"currency": str(leg.get("currency") or "").strip(),
|
|
}
|
|
)
|
|
blocks.append((title, picked, eta))
|
|
return blocks
|
|
|
|
|
|
def land_leg_fee_blocks_for_route(
|
|
quote: dict[str, Any] | None,
|
|
route: dict[str, Any] | None,
|
|
) -> list[tuple[str, list[dict[str, Any]], str]]:
|
|
"""H5 某一条备选：只收该条各段，费用名仍是 TMS 原文。"""
|
|
src = dict(route or {})
|
|
mini = dict(quote or {})
|
|
mini["selected_land_route"] = src
|
|
mini["landSegments"] = [dict(x) for x in list(src.get("legs") or []) if isinstance(x, dict)]
|
|
picked = selected_tms_fee_lines(quote, src)
|
|
if picked:
|
|
mini["fee_lines"] = picked
|
|
return land_leg_fee_blocks(mini)
|
|
|
|
|
|
def option_total(route: dict[str, Any] | None) -> str:
|
|
"""各段成本相加。加不上就空。"""
|
|
total = 0.0
|
|
ccy = ""
|
|
any_num = False
|
|
for leg in list((route or {}).get("legs") or []):
|
|
raw = leg.get("costPrice")
|
|
if raw is None:
|
|
raw = leg.get("cost_price")
|
|
if raw is None or str(raw).strip() in {"", "None", "null"}:
|
|
continue
|
|
try:
|
|
total += float(raw)
|
|
any_num = True
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if not ccy:
|
|
ccy = str(leg.get("currency") or "").strip()
|
|
if not any_num:
|
|
return "-"
|
|
text = str(int(total)) if total == int(total) else str(total)
|
|
return _money(text, ccy) or "-"
|
|
|
|
|
|
def option_eta_bits(route: dict[str, Any] | None) -> list[tuple[str, str]]:
|
|
"""每段时效，有 transitDays 才写。"""
|
|
bits: list[tuple[str, str]] = []
|
|
for i, leg in enumerate(list((route or {}).get("legs") or []), 1):
|
|
days = str(leg.get("transitDays") or leg.get("transit_days") or "").strip()
|
|
if days:
|
|
bits.append((f"第{i}段时效", days))
|
|
return bits
|
|
|
|
|
|
def project_selected_route(quote: dict[str, Any] | None, route: dict[str, Any] | None) -> dict[str, Any]:
|
|
"""
|
|
选定一条后：只留该条 landSegments，费用名按 TMS 原文，合计用成本加总。
|
|
"""
|
|
base = dict(quote or {})
|
|
src = dict(route or {})
|
|
legs = [dict(x) for x in list(src.get("legs") or [])]
|
|
base["landSegments"] = legs
|
|
lines = selected_tms_fee_lines(base, src)
|
|
if not lines:
|
|
mini = dict(base)
|
|
mini["selected_land_route"] = src
|
|
for _, fees, _eta in land_leg_fee_blocks(mini):
|
|
lines.extend(fees)
|
|
base["fee_lines"] = lines
|
|
base["total"] = option_total(src)
|
|
base["freight"] = "-"
|
|
base["other_fee"] = "-"
|
|
base["route_line"] = route_line(src)
|
|
base["selected_land_route"] = src
|
|
etas = option_eta_bits(src)
|
|
if etas:
|
|
# 时效按段原文拼接（N+2 / N+3），禁止当成小时数相加。
|
|
base["eta"] = " / ".join(d for _, d in etas)
|
|
if src:
|
|
base["cost"] = option_total(src)
|
|
return base
|