""" 陆运选项规则：运输类型、线路类别、运输分类、中港装载方式。 本文件职责：从销售原话收取标准项、校验联动、生成补问清单、判断「确定查价」意图。 联动错误只纠正那一项能选什么，不整单作废。 禁止：改六态；发明枚举；把选项当货物字段。 线程：纯函数，无共享可变状态，可在 Worker 槽调用。 """ from __future__ import annotations from typing import Any # 与后台员工管理陆运「负责线路」同一套字，禁止改字。 TRANSPORT_TYPES = ( "国内运输拼车", "国内运输整车", "跨境集拼", "跨境整车", "中港", ) DOMESTIC_ROUTES = ("国内长途/零担", "国内城配/拖车/打包") CROSS_ROUTES = ("中港/中亚/中欧", "东南亚") ROUTES_BY_TYPE = { "国内运输拼车": DOMESTIC_ROUTES, "国内运输整车": DOMESTIC_ROUTES, "跨境集拼": CROSS_ROUTES, "跨境整车": CROSS_ROUTES, "中港": CROSS_ROUTES, } CROSS_CLASSES = ( "普件，中港、东南亚常用线路/常规车型", "普件，中亚、中欧、东南亚涉及国内+国外段", "普件，中亚、中欧、东南亚涉及境外段清派等特殊要求", "急件", ) DOMESTIC_CLASSES = ( "普件，常用线路/常规车型", "普件，非常用线路/临时线路/特种车型", "急件", ) CLASSES_BY_ROUTE = { "中港/中亚/中欧": CROSS_CLASSES, "东南亚": CROSS_CLASSES, "国内长途/零担": DOMESTIC_CLASSES, "国内城配/拖车/打包": DOMESTIC_CLASSES, } HK_LOADS = ("整车", "零担/集拼") # 口语别名 → 标准运输类型。长词优先，避免「拼车」误伤「整车」。 _TYPE_ALIASES = ( ("国内运输拼车", "国内运输拼车"), ("国内运输整车", "国内运输整车"), ("国内拼车", "国内运输拼车"), ("国内整车", "国内运输整车"), ("跨境集拼", "跨境集拼"), ("跨境整车", "跨境整车"), ("中港", "中港"), ) _ROUTE_ALIASES = ( ("国内城配/拖车/打包", "国内城配/拖车/打包"), ("国内长途/零担", "国内长途/零担"), ("中港/中亚/中欧", "中港/中亚/中欧"), ("国内城配", "国内城配/拖车/打包"), ("国内长途", "国内长途/零担"), ("东南亚", "东南亚"), ) # 分类全称从长到短；「急件」单独出现才算急件，避免普件句被误收。 _CLASS_CANON = tuple( sorted( {*(CROSS_CLASSES + DOMESTIC_CLASSES)}, key=len, reverse=True, ) ) _HK_LOAD_ALIASES = ( ("零担/集拼", "零担/集拼"), ("零担", "零担/集拼"), ("集拼", "零担/集拼"), ("整车", "整车"), ) _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 canonicalize_hk_load(raw: str) -> str: """中港装载方式：整车或零担/集拼。""" text = (raw or "").strip() if not text: return "" for needle, canon in _HK_LOAD_ALIASES: if needle in text: return canon 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_type = canonicalize_transport_type(blob) found_route = canonicalize_route(blob) found_class = canonicalize_land_class(blob) found_load = canonicalize_hk_load(blob) if found_type: merged["运输类型"] = found_type if found_route: merged["线路类别"] = found_route if found_class: merged["运输分类"] = found_class if found_load: merged["中港装载方式"] = found_load 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) load = canonicalize_hk_load(merged.get("中港装载方式") or "") if load: merged["中港装载方式"] = load else: merged.pop("中港装载方式", None) def validate_land_options(opts: dict[str, str]) -> dict[str, Any]: """ 校验选项是否齐全且联动正确。 返回 ok / missing / errors。errors 每项 {field, message}，只纠正错的那一项。 中港缺装载方式进 missing，不改货物字段。 副作用：无。 """ src = {k: str(v).strip() for k, v in dict(opts or {}).items() if str(v).strip()} t = canonicalize_transport_type(src.get("运输类型") or "") or src.get("运输类型") or "" r = canonicalize_route(src.get("线路类别") or "") or src.get("线路类别") or "" c = canonicalize_land_class(src.get("运输分类") or "") or src.get("运输分类") or "" load = canonicalize_hk_load(src.get("中港装载方式") or "") or src.get("中港装载方式") or "" missing: list[str] = [] errors: list[dict[str, str]] = [] if t not in TRANSPORT_TYPES: missing.append("运输类型") t = "" if t: allowed_routes = ROUTES_BY_TYPE.get(t) or () if not r: missing.append("线路类别") elif r not in allowed_routes: errors.append( { "field": "线路类别", "message": "线路类别只能选：" + "、".join(f"「{x}」" for x in allowed_routes), } ) r = "" elif r: # 类型还没有，线路先记下但不判联动 pass if r: allowed_cls = CLASSES_BY_ROUTE.get(r) or () if not c: missing.append("运输分类") elif c not in allowed_cls: errors.append( { "field": "运输分类", "message": "运输分类只能选：" + "、".join(f"「{x}」" for x in allowed_cls), } ) c = "" elif "运输分类" not in missing and not c: missing.append("运输分类") if t == "中港": if load not in HK_LOADS: missing.append("中港装载方式") ok = not missing and not errors return { "ok": ok, "missing": missing, "errors": errors, "normalized": { "运输类型": t, "线路类别": r, "运输分类": c, "中港装载方式": load if t == "中港" else "", }, } 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 "") or src.get("运输类型") or "" r = canonicalize_route(src.get("线路类别") or "") or src.get("线路类别") or "" lines: list[str] = [] if t not in TRANSPORT_TYPES: lines.append("请一次选择：运输类型、线路类别、运输分类。") lines.append("") lines.append("运输类型：") lines.append("、".join(TRANSPORT_TYPES)) lines.append("") lines.append("线路类别（跟运输类型联动）：") lines.append("国内运输拼车 / 国内运输整车：" + "、".join(DOMESTIC_ROUTES)) lines.append("跨境集拼 / 跨境整车 / 中港：" + "、".join(CROSS_ROUTES)) lines.append("") lines.append("运输分类（跟线路类别联动）：") lines.append("中港/中亚/中欧、东南亚：" + "；".join(CROSS_CLASSES)) lines.append("国内长途/零担、国内城配/拖车/打包：" + "；".join(DOMESTIC_CLASSES)) lines.append("") lines.append("运输类型选中港时，还要选择：" + "、".join(HK_LOADS)) return "\n".join(lines).strip() # 类型已认出来：不再问运输类型 allowed_routes = ROUTES_BY_TYPE.get(t) or () if r not in allowed_routes: lines.append(f"运输类型已选「{t}」。请选择线路类别：") lines.append("、".join(allowed_routes)) route_for_class = "" else: route_for_class = r if route_for_class: allowed_cls = CLASSES_BY_ROUTE.get(route_for_class) or () if src.get("运输分类") not in allowed_cls: lines.append(f"线路类别已选「{route_for_class}」。请选择运输分类：") lines.append("；".join(allowed_cls)) elif t in TRANSPORT_TYPES: # 线路未定，把该类型下各线路的分类一并列出，方便一次粘贴 seen: set[str] = set() lines.append("运输分类（跟线路类别联动）：") for route in allowed_routes: if route in seen: continue seen.add(route) lines.append(f"{route}：" + "；".join(CLASSES_BY_ROUTE.get(route) or ())) if t == "中港" and (src.get("中港装载方式") not in HK_LOADS): lines.append("中港请再选择装载方式：" + "、".join(HK_LOADS)) 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)