""" 陆运选项规则:运输类型、线路类别、运输分类。 本文件职责:从销售原话收取标准项、校验是否在名单内、生成补问清单(含填写样例表)、判断「确定查价」意图。 三项各自必填,组合不拦截查价。拉群必须对上填写样例。运输分类句里的「中港/东南亚」不得冒充类型或线路。 禁止:改六态;发明枚举;把选项当货物字段;再要单独的中港装载方式。 线程:纯函数,无共享可变状态,可在 Worker 槽调用。 """ from __future__ import annotations from pathlib import Path from typing import Any TRANSPORT_TYPES = ( "国内运输拼车", "国内运输整车", "跨境集拼", "跨境整车", "中港整车", "中港零担/集拼", ) ROUTES = ( "国内长途/零担", "国内城配/拖车/打包", "中港/中亚/中欧", "东南亚", ) LAND_CLASSES = ( "普件(东南亚、中港常用线路/常规车型)", "普件(东南亚、中亚、中欧涉及国内+国外段)", "普件(东南亚、中亚、中欧涉及境外段清派等特殊要求)", "急件", ) # 填写样例,同时作为拉群门禁:对不上的类型+线路可以查价,但不能拉群。 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 "" 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) blob = text or "" 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, "中港装载方式": "", }, } def _class_prompt_lines() -> list[str]: lines = ["运输分类(必填):可选择回复,"] for i, name in enumerate(LAND_CLASSES, start=1): lines.append(f"{i}.{name}") return lines def option_prompt_needs_sample(*, current: dict[str, str]) -> bool: """运输类型或线路类别缺任一,就要发填写样例图。只缺分类不发。""" 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 "") return t not in TRANSPORT_TYPES or r not in ROUTES 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 lines: list[str] = [] if need_type and need_route and need_class: lines.append("请一次选择:运输类型、线路类别、运输分类") lines.append("") lines.append("运输类型(必填):可选择回复," + "、".join(TRANSPORT_TYPES)) lines.append("") lines.append("线路类别(必填):可选择回复," + "、".join(ROUTES)) lines.append("") lines.extend(_class_prompt_lines()) lines.append("") lines.append("填写样例见图。") return "\n".join(lines).strip() if need_type: lines.append("运输类型(必填):可选择回复," + "、".join(TRANSPORT_TYPES)) if need_route: if lines: lines.append("") lines.append("线路类别(必填):可选择回复," + "、".join(ROUTES)) if need_class: if lines: lines.append("") lines.extend(_class_prompt_lines()) if need_type or need_route: lines.append("") lines.append("填写样例见图。") return "\n".join(lines).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)