""" 陆运选项规则：运输类型、线路类别、运输分类。 本文件职责：从销售原话或复制填空清单收取标准项、校验是否在名单内、生成补问清单、判断「确定查价」意图。 三项各自必填，组合不拦截查价。拉群必须对上填写样例。运输分类句里的「中港/东南亚」不得冒充类型或线路。 禁止：改六态；发明枚举；把选项当货物字段；再要单独的中港装载方式。 线程：纯函数，无共享可变状态，可在 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) -> dict[str, str]: """复制清单里「字段（必填）：值」只取冒号同行，空着的不当已填。""" found: 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: continue if key == "线路类别": hit = "" for code, typ, route in ROUTE_CODES: if token == code or token.startswith(code): found["线路类别"] = route inferred_type = typ hit = code break if hit: continue decoded = _decode_option_token(key, token) if decoded: found[key] = decoded if inferred_type and not found.get("运输类型"): found["运输类型"] = inferred_type return found 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 = _parse_land_option_labels(text or "") for key in ("运输类型", "线路类别", "运输分类"): if labeled.get(key): merged[key] = labeled[key] blob = _strip_land_option_prompt(text or "") compact = blob.replace(" ", "") if compact in _ROUTE_CODE_MAP: _apply_route_code(merged, compact, fill_type=not bool(merged.get("运输类型"))) 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] elif missing_class and not missing_type and 1 <= idx <= len(LAND_CLASSES): merged["运输分类"] = LAND_CLASSES[idx - 1] 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) if found_type: merged["运输类型"] = found_type if found_route: merged["线路类别"] = found_route if found_class: merged["运输分类"] = found_class merged.pop("中港装载方式", None) return merged def _keep_canonical_options(merged: dict[str, str]) -> None: """只留能对上枚举的选项；模型胡写的「陆运」不能冒充运输类型。旧中港两项作废。""" t = canonicalize_transport_type(merged.get("运输类型") or "") if t: merged["运输类型"] = t else: merged.pop("运输类型", None) r = canonicalize_route(merged.get("线路类别") or "") if r: merged["线路类别"] = r else: merged.pop("线路类别", None) c = canonicalize_land_class(merged.get("运输分类") or "") if c: merged["运输分类"] = c else: merged.pop("运输分类", 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_prompt_lines() -> list[str]: lines = ["运输类型（必填）："] for i, name in enumerate(TRANSPORT_TYPES, start=1): lines.append(f"{i}.{name}") return lines def _route_prompt_lines() -> list[str]: lines = ["线路类别（必填）："] for code, _typ, route in ROUTE_CODES: lines.append(f"{code}{route}") return lines def _class_prompt_lines() -> list[str]: lines = ["运输分类（必填）："] for i, name in enumerate(LAND_CLASSES, start=1): lines.append(f"{i}.{name}") return lines # 编号清单已覆盖类型+线路，补问不再发填写样例图。验收脚本认这个开关。 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: """ 按已识别项生成复制填空清单。 缺哪项列哪项。销售把数字或文字填在字段冒号后面。 填写样例图不再另发。不修改 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 "") need_type = t not in TRANSPORT_TYPES need_route = r not in ROUTES need_class = c not in LAND_CLASSES if not (need_type or need_route or need_class): return "" chunks: list[list[str]] = [] if need_type: chunks.append(_type_prompt_lines()) if need_route: chunks.append(_route_prompt_lines()) if need_class: chunks.append(_class_prompt_lines()) body: list[str] = list(_OPTION_PASTE_HEAD) for i, block in enumerate(chunks): 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 return any(p in raw for p in _CONFIRM_PHRASES)