""" 空运增值服务。 本文件职责：把销售原话收成多项 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)(? 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