127 lines
4.0 KiB
Python
127 lines
4.0 KiB
Python
"""
|
|
海运查价参数：对照 TMS V2 /quote/v2/query 是否带全。
|
|
|
|
本文件职责：询价卡片出现前，确认海运十项里查出站项能拼出查价正文。
|
|
港口只带名称，不做港口字典对应；箱型箱量保持原文字段，containerItems 只透传模型投影。
|
|
贸易条款必带出站；运输分类只留在工单 facts，不进 tms_facts。
|
|
禁止：本地拆箱型；因对不上港口字典跳过 TMS；改六态。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from typing import Any
|
|
|
|
from agent.schema.field_validate import apply_quote_date_default
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _load_type(raw: str) -> str:
|
|
"""整柜/拼柜原文透传；出站可带 FCL/LCL 提示，不拒绝口语。"""
|
|
text = (raw or "").strip()
|
|
return text
|
|
|
|
|
|
def parse_container_items(facts: dict[str, Any] | None) -> list[dict[str, Any]]:
|
|
"""
|
|
只收下 DeepSeek B 已投影的 containerItems。
|
|
|
|
本地不得从「箱型箱量」字符串拆 type/quantity。
|
|
"""
|
|
src = dict(facts or {})
|
|
raw = src.get("containerItems")
|
|
if isinstance(raw, str) and raw.strip():
|
|
try:
|
|
raw = json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
return []
|
|
if not isinstance(raw, list):
|
|
return []
|
|
items: list[dict[str, Any]] = []
|
|
for row in raw:
|
|
if isinstance(row, dict):
|
|
items.append(dict(row))
|
|
return items
|
|
|
|
|
|
def assemble_sea_query(facts: dict[str, Any] | None) -> dict[str, Any]:
|
|
"""
|
|
把已采集海运字段拼成 TMS 查价参数。
|
|
|
|
不齐则 ok=False，missing 给销售补问；不建单、不出卡片。
|
|
齐了也不因没有港口代码拦网。报价日期未说则填当天。
|
|
"""
|
|
src = apply_quote_date_default(
|
|
{str(k): str(v) for k, v in dict(facts or {}).items() if v is not None}
|
|
)
|
|
origin = str(src.get("起运港") or "").strip()
|
|
dest = str(src.get("目的港") or "").strip()
|
|
cargo = str(src.get("品名") or "").strip()
|
|
qty = str(src.get("货量") or src.get("货物数量") or "").strip()
|
|
load = _load_type(str(src.get("整柜或拼柜") or ""))
|
|
box = str(src.get("箱型箱量") or "").strip()
|
|
quote_date = str(src.get("报价日期") or "").strip()
|
|
trade = str(src.get("贸易条款") or src.get("trade_terms") or src.get("tradeTerms") or "").strip()
|
|
klass = str(src.get("运输分类") or "").strip()
|
|
|
|
missing: list[str] = []
|
|
missing_display: list[str] = []
|
|
pairs = (
|
|
("起运港", "起运港", origin),
|
|
("目的港", "目的港", dest),
|
|
("品名", "品名", cargo),
|
|
("货量", "货物数量", qty),
|
|
("整柜或拼柜", "整柜或拼柜", load),
|
|
("箱型箱量", "箱型箱量", box),
|
|
("贸易条款", "贸易条款", trade),
|
|
("报价日期", "报价日期", quote_date),
|
|
)
|
|
for key, label, val in pairs:
|
|
if not val:
|
|
missing.append(key)
|
|
missing_display.append(label)
|
|
|
|
ok = not missing
|
|
out_facts = dict(src)
|
|
if origin:
|
|
out_facts["起运港"] = origin
|
|
if dest:
|
|
out_facts["目的港"] = dest
|
|
if cargo:
|
|
out_facts["品名"] = cargo
|
|
if qty:
|
|
out_facts["货量"] = qty
|
|
if load:
|
|
out_facts["整柜或拼柜"] = load
|
|
if box:
|
|
out_facts["箱型箱量"] = box
|
|
if trade:
|
|
out_facts["贸易条款"] = trade
|
|
if klass:
|
|
out_facts["运输分类"] = klass
|
|
if quote_date:
|
|
out_facts["报价日期"] = quote_date
|
|
items = parse_container_items(src)
|
|
if items:
|
|
out_facts["containerItems"] = items
|
|
|
|
tms_facts = dict(out_facts)
|
|
tms_facts.pop("运输分类", None)
|
|
logger.info(
|
|
"tms.sea.assemble ok=%s origin=%s dest=%s load=%s missing=%s",
|
|
ok,
|
|
origin or "-",
|
|
dest or "-",
|
|
load or "-",
|
|
missing,
|
|
)
|
|
return {
|
|
"ok": ok,
|
|
"missing": missing,
|
|
"missing_display": missing_display,
|
|
"facts": out_facts,
|
|
"tms_facts": tms_facts,
|
|
}
|