jillion886andCursor
cfbce2fd69
必填核对后再建单；多条线路走 H5 点选；费用只展示该线路 TMS 回包，不套用整单空运费。 Co-authored-by: Cursor <cursoragent@cursor.com>
251 lines
8.8 KiB
Python
251 lines
8.8 KiB
Python
"""
|
||
空运 TMS 线路报价：从 airOptions 拆出可选线路。
|
||
|
||
本文件职责：判断一条还是多条、拼航线（含中转）、抽出该线路费用。
|
||
选定后把起运港写回 facts。中转只用于展示，不写进报价单模板。
|
||
禁止：猜没回的机场名；把贸易条款当线路；改六态。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
from typing import Any
|
||
|
||
from agent.schema.tms_air_query import _AIRPORTS, resolve_airport_code
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 三字码 → 销售侧地名；同码多个中文时取较短的常用名。
|
||
_CODE_NAME: dict[str, str] = {}
|
||
for name, code in _AIRPORTS.items():
|
||
prev = _CODE_NAME.get(code)
|
||
if prev is None or len(name) < len(prev):
|
||
_CODE_NAME[code] = name
|
||
|
||
|
||
def parse_air_options(quote: dict[str, Any] | None) -> list[dict[str, Any]]:
|
||
"""把 quote.airOptions（list 或 JSON 串）收成字典列表。"""
|
||
src = dict(quote or {})
|
||
raw = src.get("airOptions") or src.get("air_options")
|
||
if isinstance(raw, str) and raw.strip():
|
||
try:
|
||
raw = json.loads(raw)
|
||
except json.JSONDecodeError:
|
||
logger.warning("air.routes airOptions 不是 JSON")
|
||
return []
|
||
if not isinstance(raw, list):
|
||
return []
|
||
out: list[dict[str, Any]] = []
|
||
for item in raw:
|
||
if isinstance(item, dict):
|
||
out.append(dict(item))
|
||
return out
|
||
|
||
|
||
def needs_route_pick(quote: dict[str, Any] | None) -> bool:
|
||
"""多于一条线路才进 H5 选。"""
|
||
return len(parse_air_options(quote)) > 1
|
||
|
||
|
||
def _place_name(code: str, fallback: str = "") -> str:
|
||
"""三字码能对上就用中文，否则用原文或 fallback。"""
|
||
raw = (code or "").strip()
|
||
if not raw:
|
||
return (fallback or "").strip()
|
||
mapped = _CODE_NAME.get(raw.upper())
|
||
if mapped:
|
||
return mapped
|
||
if resolve_airport_code(raw):
|
||
return _CODE_NAME.get(resolve_airport_code(raw), raw)
|
||
return raw
|
||
|
||
|
||
def route_line(option: dict[str, Any] | None, facts: dict[str, str] | None = None) -> str:
|
||
"""
|
||
销售侧航线：有中转写成 起运-中转-目的。
|
||
对不上字典就展示三字码，禁止编城市。
|
||
"""
|
||
src = dict(option or {})
|
||
facts = facts or {}
|
||
origin = _place_name(
|
||
str(src.get("originAirportCode") or src.get("origin") or ""),
|
||
str(facts.get("起运港") or ""),
|
||
)
|
||
dest = _place_name(
|
||
str(src.get("destinationAirportCode") or src.get("destination") or ""),
|
||
str(facts.get("目的港") or ""),
|
||
)
|
||
transfer = _place_name(str(src.get("transferAirportCode") or src.get("transfer") or ""))
|
||
bits = [x for x in (origin, transfer, dest) if x]
|
||
return "-".join(bits)
|
||
|
||
|
||
def _iata_code(raw: str, fallback: str = "") -> str:
|
||
"""只要三字码。已是代码就用；中文能投影就投影。禁止编假码。"""
|
||
text = (raw or "").strip()
|
||
if len(text) == 3 and text.isalpha():
|
||
return text.upper()
|
||
mapped = resolve_airport_code(text) or resolve_airport_code(fallback)
|
||
return (mapped or "").upper()
|
||
|
||
|
||
def route_codes(
|
||
option: dict[str, Any] | None,
|
||
facts: dict[str, str] | None = None,
|
||
quote: dict[str, Any] | None = None,
|
||
) -> str:
|
||
"""
|
||
价格明细/方案卡用的航线三字码：有中转写成 ZUH-CAN-CRK。
|
||
对不上就不写假中转。
|
||
"""
|
||
src = dict(option or {})
|
||
if not src:
|
||
q = dict(quote or {})
|
||
src = dict(q.get("selected_air_option") or {})
|
||
if not src:
|
||
opts = parse_air_options(q)
|
||
src = dict(opts[0]) if opts else {}
|
||
facts = facts or {}
|
||
origin = _iata_code(
|
||
str(src.get("originAirportCode") or src.get("origin") or facts.get("originCode") or ""),
|
||
str(facts.get("起运港") or ""),
|
||
)
|
||
dest = _iata_code(
|
||
str(src.get("destinationAirportCode") or src.get("destination") or facts.get("destinationCode") or ""),
|
||
str(facts.get("目的港") or ""),
|
||
)
|
||
transfer = _iata_code(
|
||
str(
|
||
src.get("transferAirportCode")
|
||
or src.get("transfer")
|
||
or (quote or {}).get("transferAirportCode")
|
||
or ""
|
||
)
|
||
)
|
||
bits = [x for x in (origin, transfer, dest) if x]
|
||
return "-".join(bits)
|
||
|
||
|
||
def scheme_route_codes(
|
||
quote: dict[str, Any] | None,
|
||
facts: dict[str, str] | None = None,
|
||
) -> str:
|
||
"""方案卡航线：空运ZUH→中转→CRK。没有代码则退回地名。"""
|
||
codes = route_codes(None, facts, quote)
|
||
if codes:
|
||
return "空运" + codes.replace("-", "→")
|
||
facts = facts or {}
|
||
origin = (facts.get("起运港") or "-").strip() or "-"
|
||
dest = (facts.get("目的港") or "-").strip() or "-"
|
||
return f"空运{origin}→{dest}"
|
||
|
||
|
||
def _fee_row(item: dict[str, Any]) -> tuple[str, str] | None:
|
||
"""一条 TMS 费用。有名称才算；金额空则用单价。禁止编费目。"""
|
||
name = str(
|
||
item.get("chargeName") or item.get("feeName") or item.get("name") or ""
|
||
).strip()
|
||
if not name:
|
||
return None
|
||
ccy = str(item.get("currency") or "").strip()
|
||
amount = item.get("amount")
|
||
if amount is None or str(amount).strip() == "":
|
||
amount = item.get("unitPrice") or item.get("unit_price")
|
||
text = str(amount or "").strip()
|
||
if ccy and text and ccy not in text:
|
||
text = f"{ccy} {text}"
|
||
return (name, text or "-")
|
||
|
||
|
||
def _rows_from_list(raw: Any) -> list[tuple[str, str]]:
|
||
"""把 TMS 费用数组收成展示行。不是列表就当没有。"""
|
||
if isinstance(raw, str) and raw.strip():
|
||
try:
|
||
raw = json.loads(raw)
|
||
except json.JSONDecodeError:
|
||
return []
|
||
if not isinstance(raw, list):
|
||
return []
|
||
out: list[tuple[str, str]] = []
|
||
for item in raw:
|
||
if isinstance(item, dict):
|
||
row = _fee_row(item)
|
||
if row:
|
||
out.append(row)
|
||
return out
|
||
|
||
|
||
def option_fee_lines(option: dict[str, Any] | None, quote: dict[str, Any] | None = None) -> list[tuple[str, str]]:
|
||
"""
|
||
该线路 TMS 回了哪些费用就列哪些。
|
||
|
||
只读本条 departureCharges / feeItems。多条线路时，整单 fee_lines 是最低价方案，
|
||
禁止套到没回费用的线路。仅一条线路时，整单 fee_lines 才算这条的明细。
|
||
"""
|
||
src = dict(option or {})
|
||
rows = _rows_from_list(src.get("departureCharges") or src.get("departure_charges"))
|
||
rows.extend(_rows_from_list(src.get("feeItems") or src.get("fee_items")))
|
||
if rows:
|
||
return rows
|
||
q = dict(quote or {})
|
||
options = parse_air_options(q)
|
||
if len(options) > 1:
|
||
return []
|
||
return _rows_from_list(q.get("fee_lines") or q.get("fee_rows") or q.get("feeItems"))
|
||
|
||
|
||
def option_total(option: dict[str, Any] | None, quote: dict[str, Any] | None = None) -> str:
|
||
"""该线路总销售价。"""
|
||
src = dict(option or {})
|
||
q = dict(quote or {})
|
||
raw = src.get("salePrice")
|
||
if raw is None or str(raw).strip() == "":
|
||
raw = q.get("total") or q.get("salePrice")
|
||
ccy = str(src.get("currency") or q.get("currency") or "").strip()
|
||
text = str(raw or "").strip() or "-"
|
||
if ccy and text != "-" and ccy not in text:
|
||
return f"{ccy} {text}"
|
||
return text
|
||
|
||
|
||
def apply_selected_option(facts: dict[str, str] | None, option: dict[str, Any] | None) -> dict[str, str]:
|
||
"""选定线路后写回起运港（及能对上的三字码）。改选覆盖。"""
|
||
merged = dict(facts or {})
|
||
src = dict(option or {})
|
||
code = str(src.get("originAirportCode") or src.get("origin") or "").strip().upper()
|
||
name = _place_name(code)
|
||
if name:
|
||
merged["起运港"] = name
|
||
if code and len(code) == 3 and code.isalpha():
|
||
merged["originCode"] = code
|
||
dest_code = str(src.get("destinationAirportCode") or "").strip().upper()
|
||
dest_name = _place_name(dest_code, str(merged.get("目的港") or ""))
|
||
if dest_name:
|
||
merged["目的港"] = dest_name
|
||
if dest_code and len(dest_code) == 3 and dest_code.isalpha():
|
||
merged["destinationCode"] = dest_code
|
||
return merged
|
||
|
||
|
||
def project_selected_quote(quote: dict[str, Any] | None, option: dict[str, Any] | None) -> dict[str, Any]:
|
||
"""把选中线路投影成价格卡用的 quote。"""
|
||
base = dict(quote or {})
|
||
src = dict(option or {})
|
||
lines = [{"name": n, "amount": a} for n, a in option_fee_lines(src, base)]
|
||
# 必须覆盖整单 fee_lines：空列表也要写回去，否则会残留别的线路空运费。
|
||
base["fee_lines"] = lines
|
||
if not lines:
|
||
base["freight"] = "-"
|
||
base["other_fee"] = "-"
|
||
total = src.get("salePrice")
|
||
if total is not None and str(total).strip() != "":
|
||
base["total"] = str(total).strip()
|
||
if src.get("currency"):
|
||
base["currency"] = src.get("currency")
|
||
if src.get("validUntil"):
|
||
base["valid_until"] = src.get("validUntil")
|
||
base["selected_air_option"] = src
|
||
base["route_line"] = route_line(src, {})
|
||
return base
|