Files
inquiry_robot/inquiry-agent/agent/schema/land_options.py
T
jillion886andCursor 9ffba70741 陆运补问把货物品名放进选项清单,并按销售原文回显已填项。
空着的运输分类不再被菜单第一项认成已填。空运报价改为货物信息加价格方案,锁舱成功回执补起运地、目的地、航司和航班日期。

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-10-14 17:39:54 +08:00

680 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
陆运选项规则:运输类型、线路类别、运输分类,以及和它们同一张清单里的货物品名。
本文件职责:从销售原话或复制填空清单收取标准项、校验三项是否在名单内、生成补问清单、判断「确定查价」意图。
三项各自必填,组合不拦截查价。货物品名六个运输类型都要,和三项同一张复制清单里填;没填完不进始发站补问。
已填项回显销售刚写的数字或文字,并附上完整选项,方便对着编号填线路类别。
拉群必须对上填写样例。运输分类句里的「中港/东南亚」不得冒充类型或线路。
禁止:改六态;发明枚举;把选项当货物字段;再要单独的中港装载方式。
线程:纯函数,无共享可变状态,可在 Worker 槽调用。
"""
from __future__ import annotations
import re
from pathlib import Path
from typing import Any
TRANSPORT_TYPES = (
"国内运输拼车",
"国内运输整车",
"跨境集拼",
"跨境整车",
"中港整车",
"中港零担/集拼",
)
ROUTES = (
"国内长途/零担",
"国内城配/拖车/打包",
"中港/中亚/中欧",
"东南亚",
)
LAND_CLASSES = (
"普件(常用线路/常规车型)",
"普件(东南亚、中港常用线路/常规车型)",
"普件(东南亚、中亚、中欧涉及国内+国外段)",
"普件(东南亚、中亚、中欧涉及境外段清派等特殊要求)",
"急件",
)
# 报价单模板码,不能当始发站/目的地。
_TEMPLATE_PORT_CODES = frozenset({"DOM", "FTL", "LTL", "INTL"})
def scrub_land_placeholder_ports(facts: dict[str, str] | None) -> dict[str, str]:
"""
丢掉报价模板里的 DOM/FTL 这类站点码。
这些不是城市或港口,当已填会让多段误出确定卡。返回新 dict,不改入参。
"""
out: dict[str, str] = {}
for key, raw in dict(facts or {}).items():
val = str(raw).strip()
if not val:
continue
if key in {"起运港", "目的港", "始发站", "目的地"} and val.upper() in _TEMPLATE_PORT_CODES:
continue
out[str(key)] = val
return out
# 复制清单里的线路编号:1.x=拼车 2.x=整车,编号比文字更短,必须先认。
ROUTE_CODES = (
("1.1", "国内运输拼车", "国内长途/零担"),
("1.2", "国内运输拼车", "国内城配/拖车/打包"),
("2.1", "国内运输整车", "国内长途/零担"),
("2.2", "国内运输整车", "国内城配/拖车/打包"),
("3.1", "跨境集拼", "中港/中亚/中欧"),
("3.2", "跨境集拼", "东南亚"),
("4.1", "跨境整车", "中港/中亚/中欧"),
("4.2", "跨境整车", "东南亚"),
("5.1", "中港整车", "中港/中亚/中欧"),
("6.1", "中港零担/集拼", "中港/中亚/中欧"),
)
_ROUTE_CODE_MAP = {code: (typ, route) for code, typ, route in ROUTE_CODES}
# 填写样例,同时作为拉群门禁:对不上的类型+线路可以查价,但不能拉群。
SAMPLE_PAIRS = (
("国内运输拼车", "国内长途/零担"),
("国内运输拼车", "国内城配/拖车/打包"),
("国内运输整车", "国内长途/零担"),
("国内运输整车", "国内城配/拖车/打包"),
("跨境集拼", "中港/中亚/中欧"),
("跨境集拼", "东南亚"),
("跨境整车", "中港/中亚/中欧"),
("跨境整车", "东南亚"),
("中港整车", "中港/中亚/中欧"),
("中港零担/集拼", "中港/中亚/中欧"),
)
# 口语别名 → 标准运输类型。长词优先。禁止把单独「中港」收成类型。
_TYPE_ALIASES = (
("国内运输拼车", "国内运输拼车"),
("国内运输整车", "国内运输整车"),
("中港零担/集拼", "中港零担/集拼"),
("中港零担", "中港零担/集拼"),
("中港集拼", "中港零担/集拼"),
("中港整车", "中港整车"),
("国内拼车", "国内运输拼车"),
("国内整车", "国内运输整车"),
("跨境集拼", "跨境集拼"),
("跨境整车", "跨境整车"),
)
_ROUTE_ALIASES = (
("国内城配/拖车/打包", "国内城配/拖车/打包"),
("国内长途/零担", "国内长途/零担"),
("中港/中亚/中欧", "中港/中亚/中欧"),
("国内城配", "国内城配/拖车/打包"),
("国内长途/零担", "国内长途/零担"),
("国内长途", "国内长途/零担"),
("国内零担", "国内长途/零担"),
("东南亚", "东南亚"),
)
# 分类全称从长到短;「急件」单独出现才算急件。
_CLASS_CANON = tuple(
sorted(LAND_CLASSES, key=len, reverse=True)
)
_CONFIRM_PHRASES = ("可以查了", "确定", "确认", "查价")
_CONFIRM_EXCLUDE = ("始发", "目的", "体积", "车型", "改成", "重量")
def canonicalize_transport_type(raw: str) -> str:
"""
把销售原话收成标准运输类型。
参数:任意一句。返回 TRANSPORT_TYPES 之一或空串。
只说「中港」不算选完,避免分类句里的中港把类型改掉。
副作用:无。先匹配更长别名。
"""
text = (raw or "").strip()
if not text:
return ""
for needle, canon in _TYPE_ALIASES:
if needle in text:
return canon
return ""
def canonicalize_route(raw: str) -> str:
"""把原话收成标准线路类别;对不上返回空串。不把单独「中港」收成中港/中亚/中欧。"""
text = (raw or "").strip()
if not text:
return ""
for needle, canon in _ROUTE_ALIASES:
if needle in text:
return canon
return ""
def canonicalize_land_class(raw: str) -> str:
"""
把原话收成标准运输分类。
先匹配完整分类句;整句只有「急件」才收急件,避免普件描述被截成急件。
"""
text = (raw or "").strip()
if not text:
return ""
for canon in _CLASS_CANON:
if canon == "急件":
continue
if canon in text:
return canon
stripped = text.replace("运输分类", "").replace(":", "").replace(":", "").strip()
if stripped == "急件" or text.strip() == "急件":
return "急件"
if "急件" in text and "普件" not in text:
return "急件"
return ""
def sample_pair_allowed(land_type: str, route: str) -> bool:
"""
类型+线路是否在填写样例里。
查价不调用本函数;点「拉产品进群协同」时必须为真,否则不建群。
入参先收成标准项。副作用:无。
"""
t = canonicalize_transport_type(land_type) or (land_type or "").strip()
r = canonicalize_route(route) or (route or "").strip()
return (t, r) in SAMPLE_PAIRS
def canonicalize_hk_load(raw: str) -> str:
"""
旧中港装载方式不再收取。
保留函数名避免旧调用处报缺符号;恒返回空串。新单用运输类型「中港整车」「中港零担/集拼」。
"""
_ = raw
return ""
_PASTE_LABEL = re.compile(
r"(运输类型|线路类别|运输分类)(?:(必填))?[ \t]*[::][ \t]*([^\n]*)"
)
# 1.国内拼车 / 1.1国内长途 都是清单选项行,不能当销售已答。
_MENU_LINE = re.compile(r"^\s*(?:\d+\.\d+|\d+\.)")
_STEP_LINE = re.compile(r"^\s*Step[12]\s*:", re.I)
_COPY_FENCE = re.compile(r"^\s*-{2,}请复制以下内容-{2,}\s*$")
def _apply_route_code(merged: dict[str, str], raw: str, *, fill_type: bool) -> bool:
"""编号 1.1~6.1 收成线路;类型还空着时按编号补上。"""
token = (raw or "").strip()
if not token:
return False
code = ""
for key in sorted(_ROUTE_CODE_MAP, key=len, reverse=True):
if token == key or token.startswith(key):
code = key
break
if not code:
return False
typ, route = _ROUTE_CODE_MAP[code]
merged["线路类别"] = route
if fill_type and not str(merged.get("运输类型") or "").strip():
merged["运输类型"] = typ
return True
def _decode_option_token(key: str, raw: str) -> str:
"""冒号后的数字或文字收成标准项。认不到返回空串。"""
token = (raw or "").strip()
if not token:
return ""
if key == "运输类型":
if token.isdigit():
idx = int(token)
if 1 <= idx <= len(TRANSPORT_TYPES):
return TRANSPORT_TYPES[idx - 1]
return canonicalize_transport_type(token)
if key == "线路类别":
for code, _typ, route in ROUTE_CODES:
if token == code or token.startswith(code):
return route
return canonicalize_route(token)
if key == "运输分类":
if token.isdigit():
idx = int(token)
if 1 <= idx <= len(LAND_CLASSES):
return LAND_CLASSES[idx - 1]
return canonicalize_land_class(token)
return ""
def _parse_land_option_labels(text: str) -> tuple[dict[str, str], dict[str, str]]:
"""
复制清单里「字段(必填):值」只取冒号同行,空着的不当已填。
返回标准项,以及销售冒号后原文,供下一张清单原样回显。
"""
found: dict[str, str] = {}
shown: dict[str, str] = {}
inferred_type = ""
for match in _PASTE_LABEL.finditer(text or ""):
key = match.group(1)
token = (match.group(2) or "").strip()
if not token or _CARGO_PLACEHOLDER.match(token):
continue
if key == "线路类别":
hit = ""
for code, typ, route in ROUTE_CODES:
if token == code or token.startswith(code):
found["线路类别"] = route
shown["线路类别"] = token
inferred_type = typ
hit = code
break
if hit:
continue
decoded = _decode_option_token(key, token)
if decoded:
found[key] = decoded
shown[key] = token
if inferred_type and not found.get("运输类型"):
# 编号带出的类型不是销售本行写下的字。已有回显时不要改成标准名称。
found["运输类型"] = inferred_type
shown.setdefault("运输类型", "")
return found, shown
def _strip_land_option_prompt(text: str) -> str:
"""
丢掉补问清单的 Step 和编号选项行,避免未填清单被收成已选。
整句只有一个编号(如 1.1)要留下,那是销售在答题。
"""
raw = (text or "").strip()
if not raw:
return ""
compact = raw.replace(" ", "")
if compact in _ROUTE_CODE_MAP or compact.isdigit():
return raw
kept: list[str] = []
for line in raw.splitlines():
if _STEP_LINE.match(line):
continue
if _COPY_FENCE.match(line):
continue
if _PASTE_LABEL.match(line):
continue
if _MENU_LINE.match(line):
continue
kept.append(line)
return "\n".join(kept).strip()
def harvest_land_options(text: str, current: dict[str, str] | None = None) -> dict[str, str]:
"""
从一句原话收取陆运选项,合并到已有值上。
先收复制清单冒号后的数字/编号/文字,再收口语。
未填的补问清单本身不能当成已选。
先收运输分类并把命中句从原文挖掉,再收类型/线路,避免分类里的东南亚、中港冒充选项。
已有非空项默认保留;本句明确写出另一标准项时才覆盖。
已有值对不上标准枚举则丢掉。不改货物字段。
回显键只在本句真正改了该项时更新:复制清单留冒号后原文,口语留标准名称。
返回新 dict,不修改 current。
"""
merged = {
k: str(v).strip()
for k, v in dict(current or {}).items()
if str(v).strip()
}
_keep_canonical_options(merged)
labeled, shown = _parse_land_option_labels(text or "")
for key in ("运输类型", "线路类别", "运输分类"):
if not labeled.get(key):
continue
previous = str(merged.get(key) or "").strip()
merged[key] = labeled[key]
token = str(shown.get(key) or "").strip()
if token:
_remember_show(merged, key, token, labeled[key])
elif previous != labeled[key] or not str(merged.get(land_show_key(key)) or "").strip():
_remember_show(merged, key, labeled[key], labeled[key])
blob = _strip_land_option_prompt(text or "")
compact = blob.replace(" ", "")
if compact in _ROUTE_CODE_MAP:
had_type = bool(merged.get("运输类型"))
_apply_route_code(merged, compact, fill_type=not had_type)
_remember_show(merged, "线路类别", compact, str(merged.get("线路类别") or ""))
if not had_type and merged.get("运输类型"):
_remember_show(merged, "运输类型", merged["运输类型"], merged["运输类型"])
merged.pop("中港装载方式", None)
return merged
if compact.isdigit() and len(compact) == 1:
idx = int(compact)
missing_type = not merged.get("运输类型")
missing_class = not merged.get("运输分类")
if missing_type and not missing_class and 1 <= idx <= len(TRANSPORT_TYPES):
merged["运输类型"] = TRANSPORT_TYPES[idx - 1]
_remember_show(merged, "运输类型", compact, merged["运输类型"])
elif missing_class and not missing_type and 1 <= idx <= len(LAND_CLASSES):
merged["运输分类"] = LAND_CLASSES[idx - 1]
_remember_show(merged, "运输分类", compact, merged["运输分类"])
merged.pop("中港装载方式", None)
return merged
found_class = canonicalize_land_class(blob)
remainder = blob.replace(found_class, " ") if found_class else blob
found_type = canonicalize_transport_type(remainder)
found_route = canonicalize_route(remainder)
_assign_spoken_option(merged, "运输类型", found_type)
_assign_spoken_option(merged, "线路类别", found_route)
_assign_spoken_option(merged, "运输分类", found_class)
merged.pop("中港装载方式", None)
return merged
def _assign_spoken_option(merged: dict[str, str], key: str, canonical: str) -> None:
"""口语命中才覆盖。标准名没变时保留销售上次写的数字回显。"""
if not canonical:
return
previous = str(merged.get(key) or "").strip()
merged[key] = canonical
if previous != canonical or not str(merged.get(land_show_key(key)) or "").strip():
_remember_show(merged, key, canonical, canonical)
# 回显用:销售写「5」就留 5,写「中港整车」就留这句。只进会话,建单前丢掉。
LAND_SHOW_PREFIX = "_land_show_"
_CARGO_PLACEHOLDER = re.compile(r"^[((]\s*如\s*[::]")
def land_show_key(field: str) -> str:
"""选项回显键。不进 TMS,也不出现在核对卡。"""
return LAND_SHOW_PREFIX + field
def drop_land_prompt_echo(facts: dict[str, Any] | None) -> dict[str, str]:
"""建单前去掉回显键。返回新 dict,不改入参。"""
out: dict[str, str] = {}
for key, raw in dict(facts or {}).items():
name = str(key)
if name.startswith(LAND_SHOW_PREFIX):
continue
val = str(raw).strip()
if val:
out[name] = val
return out
def blank_paste_option_fields(text: str) -> set[str]:
"""
复制清单里出现了选项标签、但冒号后没写值的那些项。
编号菜单和模型不能把「1.普件…」抄成已填。不是复制清单时返回空,口语「急件」仍可收。
"""
raw = text or ""
if "请复制以下内容" not in raw and "Step1" not in raw:
return set()
seen: set[str] = set()
filled: set[str] = set()
for match in _PASTE_LABEL.finditer(raw):
key = match.group(1)
token = (match.group(2) or "").strip()
seen.add(key)
if token and not _CARGO_PLACEHOLDER.match(token):
filled.add(key)
return seen - filled
def drop_blank_paste_options(facts: dict[str, str] | None, text: str) -> dict[str, str]:
"""
复制清单里空着的运输类型、线路类别、运输分类,从本句结果里拿掉。
模型若按菜单猜了一项,也不留。返回新 dict,不改入参。
"""
out = {
str(k): str(v).strip()
for k, v in dict(facts or {}).items()
if str(v).strip()
}
for key in blank_paste_option_fields(text):
out.pop(key, None)
out.pop(land_show_key(key), None)
return out
def land_cargo_name(facts: dict[str, Any] | None) -> str:
"""
陆运货物品名。六个运输类型都要。
示例行「(如:衣服)」原样贴回不算已填。没有则空串。
"""
src = facts or {}
name = str(src.get("品名") or src.get("货物品名") or "").strip()
if not name or _CARGO_PLACEHOLDER.match(name):
return ""
return name
def _remember_show(merged: dict[str, str], key: str, shown: str, canonical: str) -> None:
"""记下销售刚写的数字或文字。超长句子改回标准名称,避免把整段需求贴进回显。"""
token = (shown or "").strip()
if not token or "\n" in token or len(token) > 80:
token = canonical
if token:
merged[land_show_key(key)] = token
def _keep_canonical_options(merged: dict[str, str]) -> None:
"""只留能对上枚举的选项;模型胡写的「陆运」不能冒充运输类型。旧中港两项作废。"""
t = canonicalize_transport_type(merged.get("运输类型") or "")
if t:
merged["运输类型"] = t
else:
merged.pop("运输类型", None)
merged.pop(land_show_key("运输类型"), None)
r = canonicalize_route(merged.get("线路类别") or "")
if r:
merged["线路类别"] = r
else:
merged.pop("线路类别", None)
merged.pop(land_show_key("线路类别"), None)
c = canonicalize_land_class(merged.get("运输分类") or "")
if c:
merged["运输分类"] = c
else:
merged.pop("运输分类", None)
merged.pop(land_show_key("运输分类"), None)
merged.pop("中港装载方式", None)
def validate_land_options(opts: dict[str, str]) -> dict[str, Any]:
"""
校验三项是否都在名单内。组合不拦截。
返回 ok / missing / errors。对不上名单的项当缺项,不写联动纠正句。
副作用:无。
"""
src = {k: str(v).strip() for k, v in dict(opts or {}).items() if str(v).strip()}
t = canonicalize_transport_type(src.get("运输类型") or "")
r = canonicalize_route(src.get("线路类别") or "")
c = canonicalize_land_class(src.get("运输分类") or "")
missing: list[str] = []
if t not in TRANSPORT_TYPES:
missing.append("运输类型")
t = ""
if r not in ROUTES:
missing.append("线路类别")
r = ""
if c not in LAND_CLASSES:
missing.append("运输分类")
c = ""
ok = not missing
return {
"ok": ok,
"missing": missing,
"errors": [],
"normalized": {
"运输类型": t,
"线路类别": r,
"运输分类": c,
"中港装载方式": "",
},
}
_OPTION_PASTE_HEAD = (
"Step1: 请复制以下内容,粘贴到输入框;",
"Step2:在粘贴内容的每个字段的冒号后,回复数字或文字,告诉我对应字段的值:",
"---请复制以下内容---",
)
def _type_menu_lines() -> list[str]:
return [f"{i}.{name}" for i, name in enumerate(TRANSPORT_TYPES, start=1)]
def _route_menu_lines() -> list[str]:
return [f"{code}{route}" for code, _typ, route in ROUTE_CODES]
def _class_menu_lines() -> list[str]:
return [f"{i}.{name}" for i, name in enumerate(LAND_CLASSES, start=1)]
def _type_prompt_lines() -> list[str]:
return ["运输类型(必填):", *_type_menu_lines()]
def _route_prompt_lines() -> list[str]:
return ["线路类别(必填):", *_route_menu_lines()]
def _class_prompt_lines() -> list[str]:
return ["运输分类(必填):", *_class_menu_lines()]
def _cargo_prompt_lines() -> list[str]:
"""货物品名没有编号。示例单独一行,原样贴回不算已填。"""
return ["货物品名(必填):", "(如:衣服)"]
def _shown_option(current: dict[str, str], key: str, canonical: str) -> str:
"""回显销售上次写的数字或文字;没有则用标准名称。"""
raw = str(current.get(land_show_key(key)) or "").strip()
return raw or canonical
# 编号清单已覆盖类型+线路,补问不再发填写样例图。验收脚本认这个开关。
SAMPLE_IMAGE_DISABLED = True
def option_prompt_needs_sample(*, current: dict[str, str]) -> bool:
"""
填写样例图已停发。编号清单已经覆盖类型+线路搭配,不再另出图。
保留函数名避免旧调用处报缺符号;恒为假。current 不使用。
"""
_ = current
return False
def sample_image_path() -> Path:
"""填写样例表 PNG。和源码放一起,随 agent 发布包走。"""
return Path(__file__).resolve().parent.parent / "assets" / "land_sample_pairs.png"
def sample_fallback_text() -> str:
"""
图片发不出去时的文字兜底。
一行一种搭配,不再画框线。
"""
lines = ["填写样例"]
for i, (typ, route) in enumerate(SAMPLE_PAIRS, start=1):
lines.append(f"{i}.{typ} + {route}")
return "\n".join(lines)
def _sample_table() -> str:
"""兼容旧名:补问正文不再内嵌样例,只作图片失败兜底。"""
return sample_fallback_text()
def option_prompt(*, current: dict[str, str]) -> str:
"""
按已识别项生成复制填空清单。
运输类型、线路类别、运输分类、货物品名同一张清单。缺的留在复制框里。
已经填过的排在 Step1 上面:保留销售写的数字或文字,并附上完整选项。
货物品名没有选项,填了只回显一行。四个都齐了返回空串。
填写样例图不再另发。不修改 current。
"""
src = {k: str(v).strip() for k, v in dict(current or {}).items() if str(v).strip()}
t = canonicalize_transport_type(src.get("运输类型") or "")
r = canonicalize_route(src.get("线路类别") or "")
c = canonicalize_land_class(src.get("运输分类") or "")
cargo = land_cargo_name(src)
filled: list[list[str]] = []
missing: list[list[str]] = []
if t in TRANSPORT_TYPES:
filled.append(
[f"运输类型(必填):{_shown_option(src, '运输类型', t)}", *_type_menu_lines()]
)
else:
missing.append(_type_prompt_lines())
if r in ROUTES:
filled.append(
[f"线路类别(必填):{_shown_option(src, '线路类别', r)}", *_route_menu_lines()]
)
else:
missing.append(_route_prompt_lines())
if c in LAND_CLASSES:
filled.append(
[f"运输分类(必填):{_shown_option(src, '运输分类', c)}", *_class_menu_lines()]
)
else:
missing.append(_class_prompt_lines())
if cargo:
filled.append([f"货物品名(必填):{cargo}"])
else:
missing.append(_cargo_prompt_lines())
if not missing:
return ""
body: list[str] = []
for i, block in enumerate(filled):
if i:
body.append("")
body.extend(block)
if filled:
body.append("")
body.extend(_OPTION_PASTE_HEAD)
for i, block in enumerate(missing):
if i:
body.append("")
body.extend(block)
return "\n".join(body).strip()
def looks_like_confirm(text: str) -> bool:
"""
判断是否在表示确定查价。
「确定」「确认」「查价」「可以查了」及相近短句才算。
长询价里「需确认/请确认」不算点核对卡,避免误回「核对已过期」。
同一句像在改始发/目的/体积/车型/「改成」则不当确定。
纯函数,无副作用。
"""
raw = (text or "").strip()
if not raw:
return False
if any(x in raw for x in _CONFIRM_EXCLUDE):
return False
# 超过约 20 字一定是需求正文,不是「确定」短答。
if len(raw) > 20:
return False
compact = re.sub(r"[\s。!?!?,,、]+", "", raw)
if not compact:
return False
return any(p in compact for p in _CONFIRM_PHRASES)