268 lines
9.4 KiB
Python
268 lines
9.4 KiB
Python
"""
|
||
空运增值服务。
|
||
|
||
本文件职责：把销售原话收成多项 added_services，供确认、工单和 TMS 的 addedServices 使用。
|
||
12 个服务词只要完整出现就记；名单外的必须写在「增值服务」后面才记，并保留原词。
|
||
包装类型/包装方式这一栏里的「托盘」只算包装，不抄进增值服务。
|
||
「增值服务：」同一行为空时，不把下一行收进来，也不沿用上一句已经记下的服务。
|
||
不挡查价。不改六态。海运、陆运不走这里。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
|
||
from agent.schema.field_validate import _PACK_WORDS, _without_clarify_examples
|
||
|
||
# 长词在前，避免「卡转低消」再拆出「卡转」。
|
||
_CN_TERMS = (
|
||
"卡转低消",
|
||
"前置仓",
|
||
"托盘",
|
||
"带电",
|
||
"卡转",
|
||
"备装",
|
||
"分泡",
|
||
"温控",
|
||
"快件",
|
||
)
|
||
_CN_RE = re.compile("|".join(re.escape(word) for word in _CN_TERMS))
|
||
_MUST_GO_RE = re.compile(r"(?i)MUST\s*GO")
|
||
_DG_RE = re.compile(r"(?i)(?<![A-Za-z])DG(?![A-Za-z])")
|
||
_TC_RE = re.compile(r"(?i)(?<![A-Za-z])TC(?![A-Za-z])")
|
||
# 冒号后只吃同一行。\s 会跨行，空着的增值服务会把下一行「报价日期（非必填）：」吞进来。
|
||
_LABEL_RE = re.compile(
|
||
r"增值服务(?:（非必填）|\(非必填\))?[^\S\n]*[:：][^\S\n]*([^\n]*)"
|
||
)
|
||
# 包装类型/包装方式这一栏里的托盘只算包装。确认卡上增值服务留空时不能再抄一份过去。
|
||
_PACKAGE_PALLET_RE = re.compile(
|
||
r"包装(?:类型|方式)?(?:（(?:非)?必填）|\((?:非)?必填\))?[^\S\n]*[:：][^\S\n]*托盘"
|
||
)
|
||
_SPLIT_RE = re.compile(r"[、，,]")
|
||
_NEG_PREFIX = frozenset("不无非")
|
||
_FILLER = re.compile(r"[、，,\s和及与:：]+")
|
||
|
||
|
||
def split_added_services(raw: str) -> list[str]:
|
||
"""
|
||
把一栏增值服务拆成多项。
|
||
|
||
顿号、中文逗号、英文逗号都是分隔。空格不拆，MUST GO 仍是一项。
|
||
副作用：无。
|
||
"""
|
||
text = (raw or "").strip()
|
||
if not text:
|
||
return []
|
||
out: list[str] = []
|
||
seen: set[str] = set()
|
||
for part in _SPLIT_RE.split(text):
|
||
token = part.strip().strip("。；; ")
|
||
key = token.casefold()
|
||
if not token or key in seen:
|
||
continue
|
||
seen.add(key)
|
||
out.append(token)
|
||
return out
|
||
|
||
|
||
def join_added_services(items: list[str]) -> str:
|
||
"""多项用顿号连成销售看到的一行。空则空串。"""
|
||
return "、".join(split_added_services("、".join(items)))
|
||
|
||
|
||
def harvest_added_services(text: str) -> list[str]:
|
||
"""
|
||
从当前原话按出现顺序抽出增值服务。
|
||
|
||
补问示例行「(如：托盘、…)」整行不看，避免清单被当成已填。
|
||
「不带电」不算带电；「卡转低消」不算再加一项卡转。
|
||
副作用：无。
|
||
"""
|
||
raw = _without_clarify_examples(text or "")
|
||
events: list[tuple[int, str]] = []
|
||
# 包装栏里的托盘已在 _canonical_events 排除，这里只收正文里真正的服务词。
|
||
events.extend(_canonical_events(raw))
|
||
for match in _LABEL_RE.finditer(raw):
|
||
value = (match.group(1) or "").strip()
|
||
if not value or value.startswith("(如") or value.startswith("（如"):
|
||
continue
|
||
base = match.start(1)
|
||
cursor = 0
|
||
for part in _SPLIT_RE.split(value):
|
||
piece = part.strip().strip("。；; ")
|
||
if not piece:
|
||
continue
|
||
at = value.find(part, cursor)
|
||
if at < 0:
|
||
at = cursor
|
||
cursor = at + len(part)
|
||
extra = _outside_list_text(piece)
|
||
if extra:
|
||
events.append((base + at + max(piece.find(extra), 0), extra))
|
||
events.sort(key=lambda item: item[0])
|
||
return _dedupe([word for _pos, word in events])
|
||
|
||
|
||
def apply_air_value_added(
|
||
text: str,
|
||
facts: dict[str, str] | None,
|
||
previous: str = "",
|
||
) -> dict[str, str]:
|
||
"""
|
||
把本句增值服务并进事实，并避免托盘盖住已经说出的其他包装。
|
||
|
||
previous 是上一句已经记下的 added_services。本句没提服务时保留旧值。
|
||
只应在空运调用。副作用：返回新字典，不改入参。
|
||
"""
|
||
merged = dict(facts or {})
|
||
found = harvest_added_services(text)
|
||
labeled = _labeled_added_value(text)
|
||
# 销售把「增值服务：」留空：这一栏就是空的。
|
||
# 不沿用上一句，也不把包装类型里的托盘、下一行报价日期算进来。
|
||
# 本句没写这一栏、也没提到服务时，才保留已经记下的旧值。
|
||
if labeled == "":
|
||
items = found
|
||
elif found:
|
||
items = _dedupe(split_added_services(previous) + found)
|
||
else:
|
||
items = _dedupe(
|
||
split_added_services(previous)
|
||
+ split_added_services(str(merged.get("added_services") or ""))
|
||
)
|
||
if items:
|
||
merged["added_services"] = "、".join(items)
|
||
elif not str(merged.get("added_services") or "").strip():
|
||
merged.pop("added_services", None)
|
||
_keep_stated_package(text, merged)
|
||
return merged
|
||
|
||
|
||
def _canonical_events(raw: str) -> list[tuple[int, str]]:
|
||
"""按原文位置列出 12 个标准词。重叠区间只留更长的那一项。"""
|
||
found: list[tuple[int, int, str]] = []
|
||
package_pallets = _package_pallet_spans(raw)
|
||
for match in _CN_RE.finditer(raw):
|
||
word = match.group(0)
|
||
if word == "带电" and match.start() > 0 and raw[match.start() - 1] in _NEG_PREFIX:
|
||
continue
|
||
if word == "托盘" and _span_inside(match.start(), match.end(), package_pallets):
|
||
continue
|
||
found.append((match.start(), match.end(), word))
|
||
for match in _MUST_GO_RE.finditer(raw):
|
||
found.append((match.start(), match.end(), "MUST GO"))
|
||
for match in _DG_RE.finditer(raw):
|
||
found.append((match.start(), match.end(), "DG"))
|
||
for match in _TC_RE.finditer(raw):
|
||
found.append((match.start(), match.end(), "TC"))
|
||
found.sort(key=lambda item: (item[0], -(item[1] - item[0])))
|
||
accepted: list[tuple[int, int]] = []
|
||
events: list[tuple[int, str]] = []
|
||
for start, end, word in found:
|
||
if any(not (end <= left or start >= right) for left, right in accepted):
|
||
continue
|
||
accepted.append((start, end))
|
||
events.append((start, word))
|
||
return events
|
||
|
||
|
||
def _outside_list_text(piece: str) -> str:
|
||
"""
|
||
增值服务标签里、12 个词没有覆盖到的原话。
|
||
|
||
「托盘」本身会被标准词扫到，这里不重复收。
|
||
「打木架和带电」只把打木架留下来，带电仍由标准词按出现位置记。
|
||
"""
|
||
spans = _canonical_events(piece)
|
||
if not spans:
|
||
return piece.strip()
|
||
chars = list(piece)
|
||
for start, word in spans:
|
||
# _canonical_events 只有起点。用标准词长度擦掉；MUST GO 用原文长度。
|
||
end = start + _span_len(piece, start, word)
|
||
for index in range(start, min(end, len(chars))):
|
||
chars[index] = " "
|
||
rest = _FILLER.sub("", "".join(chars))
|
||
return rest.strip()
|
||
|
||
|
||
def _span_len(piece: str, start: int, word: str) -> int:
|
||
"""标准词在原文里占几位。MUST GO 中间空格可有可无。"""
|
||
if word == "MUST GO":
|
||
match = _MUST_GO_RE.match(piece, start)
|
||
if match:
|
||
return match.end() - match.start()
|
||
if word in {"DG", "TC"}:
|
||
return 2
|
||
return len(word)
|
||
|
||
|
||
def _labeled_added_value(text: str) -> str | None:
|
||
"""
|
||
原话里有没有单独的「增值服务」栏。
|
||
|
||
没写这一栏返回 None。写了但冒号后同一行为空，返回空串。
|
||
只看同一行，避免空栏把下一行报价日期收进来。
|
||
"""
|
||
raw = _without_clarify_examples(text or "")
|
||
seen = False
|
||
value = ""
|
||
for match in _LABEL_RE.finditer(raw):
|
||
seen = True
|
||
piece = (match.group(1) or "").strip().strip("。；; ")
|
||
if not piece or piece.startswith("(如") or piece.startswith("（如"):
|
||
continue
|
||
value = piece
|
||
if not seen:
|
||
return None
|
||
return value
|
||
|
||
|
||
def _package_pallet_spans(raw: str) -> list[tuple[int, int]]:
|
||
"""「包装类型：托盘」里那个托盘的起止。只算包装，不算增值服务。"""
|
||
spans: list[tuple[int, int]] = []
|
||
for match in _PACKAGE_PALLET_RE.finditer(raw or ""):
|
||
end = match.end()
|
||
spans.append((end - len("托盘"), end))
|
||
return spans
|
||
|
||
|
||
def _span_inside(start: int, end: int, spans: list[tuple[int, int]]) -> bool:
|
||
"""当前词是否落在给定区间里。"""
|
||
return any(start >= left and end <= right for left, right in spans)
|
||
|
||
|
||
def _keep_stated_package(text: str, facts: dict[str, str]) -> None:
|
||
"""
|
||
同一句里还有纸箱、木箱、卡板、散货时，包装类型保持那个词。
|
||
|
||
包装栏里的托盘不进增值服务，只在没另说包装时由机读记成包装类型。
|
||
"""
|
||
raw = _without_clarify_examples(text or "")
|
||
if "托盘" not in raw:
|
||
return
|
||
last = ""
|
||
last_pos = -1
|
||
for word in _PACK_WORDS:
|
||
if word == "托盘":
|
||
continue
|
||
pos = raw.rfind(word)
|
||
if pos > last_pos:
|
||
last_pos = pos
|
||
last = word
|
||
if last:
|
||
facts["包装方式"] = last
|
||
|
||
|
||
def _dedupe(items: list[str]) -> list[str]:
|
||
"""按第一次出现保留，英文忽略大小写。"""
|
||
out: list[str] = []
|
||
seen: set[str] = set()
|
||
for item in items:
|
||
token = (item or "").strip()
|
||
key = token.casefold()
|
||
if not token or key in seen:
|
||
continue
|
||
seen.add(key)
|
||
out.append(token)
|
||
return out
|