269 lines
9.7 KiB
Python
269 lines
9.7 KiB
Python
"""
|
||
私聊建单前的违禁品拦截。
|
||
|
||
本文件职责：用固定词表判断当前这句话、尚未改口的原话，以及品名、起运地、备注
|
||
是否涉及真家伙（枪支、弹药、爆炸物、毒品）。命中则不能建单。
|
||
群聊、已经建成的工单不在这里判断，由调用方避开。
|
||
调用：Worker 里的私聊文字流程，在出补问或建单之前。只做字符串匹配，不调模型。
|
||
禁止：把水枪、玩具枪、仿真枪、气枪、烟花、礼炮、鞭炮算进去；禁止拦化学品、电池、油品。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from typing import Any
|
||
|
||
# 销售私聊里只回这一句。不要再追问缺项，也不要带工单号。
|
||
BLOCK_REPLY = "涉及危险品，无法发起报价。"
|
||
|
||
# 先圈出放行说法。词表里的「枪支」若落在这些说法内部，不算命中。
|
||
_ALLOW = (
|
||
"玩具枪",
|
||
"水枪",
|
||
"仿真枪",
|
||
"气枪",
|
||
"烟花",
|
||
"礼炮",
|
||
"鞭炮",
|
||
"炮仗",
|
||
)
|
||
|
||
# 真家伙。不用单字「枪」「炮」「弹」，避免误伤水枪和礼炮。
|
||
_BLOCK = (
|
||
"汽油弹",
|
||
"燃烧弹",
|
||
"原子弹",
|
||
"手榴弹",
|
||
"导火索",
|
||
"甲基苯丙胺",
|
||
"氯胺酮",
|
||
"高射炮",
|
||
"榴弹炮",
|
||
"冲锋枪",
|
||
"狙击枪",
|
||
"霰弹枪",
|
||
"火药枪",
|
||
"摇头丸",
|
||
"海洛因",
|
||
"可卡因",
|
||
"炸弹",
|
||
"炸药",
|
||
"爆破",
|
||
"雷管",
|
||
"手雷",
|
||
"地雷",
|
||
"导弹",
|
||
"核弹",
|
||
"枪支",
|
||
"枪械",
|
||
"手枪",
|
||
"步枪",
|
||
"机枪",
|
||
"猎枪",
|
||
"子弹",
|
||
"弹药",
|
||
"弹头",
|
||
"大炮",
|
||
"火炮",
|
||
"毒品",
|
||
"冰毒",
|
||
"大麻",
|
||
"鸦片",
|
||
"吗啡",
|
||
"K粉",
|
||
)
|
||
|
||
# 改口动词。多段原文里的「品名：衣服」不能走这里，否则会抄到每一段。
|
||
_FIELD_EDIT = re.compile(
|
||
r"(货物品名|品名|起运地|起运港|始发站|备注)\s*(?:改为|改成|是|为)\s*(\S+)"
|
||
)
|
||
# 单票里销售按清单重写「品名：服装」。多段「段1 / 段2」整单不靠这个改字段。
|
||
_LABEL_EDIT = re.compile(
|
||
r"(货物品名|品名|起运地|起运港|始发站|备注)\s*[:：]\s*(\S+)"
|
||
)
|
||
_MULTI_SHEET = re.compile(r"段\s*\d|第\s*[一二三四五六七八九\d]\s*段")
|
||
# 口语改口：运送衣服、改成衣服、换成衣服、炸弹换衣服。不认「换一下」「换个港口」。
|
||
_SPOKEN_CARGO = re.compile(
|
||
r"(?:运送|托运|发运|改成|改为|换成|换为)\s*([^\s，,。；;、请]{1,16})"
|
||
r"|换(?!成|为|个|一下)([^\s，,。；;、请]{1,16})"
|
||
)
|
||
_CARGO_TAIL = re.compile(r"(请报价|报价|一下|吧|了)+$")
|
||
_CARGO_SKIP = frozenset({"一下", "个", "报价", "价", "询价", "港口", "地方"})
|
||
|
||
# 销售侧说法写回内部键。起运地在主账里仍是起运港。
|
||
_WRITE_KEYS = {
|
||
"品名": ("品名", "货物品名"),
|
||
"货物品名": ("品名", "货物品名"),
|
||
"起运地": ("起运港", "起运地", "始发站"),
|
||
"起运港": ("起运港", "起运地", "始发站"),
|
||
"始发站": ("起运港", "起运地", "始发站"),
|
||
"备注": ("备注",),
|
||
}
|
||
|
||
# 建单前要看的当前字段。没写进这几项、但原话里出现过的，靠 sensitive_hold 记住。
|
||
_SCAN_KEYS = ("品名", "货物品名", "起运港", "起运地", "始发站", "备注")
|
||
|
||
# 新开一票的说法。命中且本句没有违禁词时，丢掉上一句没改口的 hold。
|
||
_FRESH_INQUIRY = re.compile(
|
||
r"(查一下|帮我查|询一下|询个价|询价|请报价|报一下价|报个价|费用)"
|
||
)
|
||
# 只补量尺、件数、包装：不算新询价，旧 hold 继续有效。
|
||
_MEASURE_SUPPLEMENT = re.compile(
|
||
r"(?:"
|
||
r"重量|毛重|体积|件数|货量|包装(?:方式|类型)?"
|
||
r")\s*[:：]?\s*\d"
|
||
r"|\d+\s*(?:kgs?|kg|公斤|千克|吨|t|cbm|方|件|箱|托)",
|
||
re.I,
|
||
)
|
||
|
||
|
||
def looks_like_fresh_inquiry(text: str) -> bool:
|
||
"""
|
||
当前句是不是另开一票，而不是接着补上一票的字段。
|
||
|
||
「查一下目的地吉隆坡的费用」算新开；「重量100KGS」不算。
|
||
用来在本句已无违禁词时清掉 sticky hold，避免正常询价被上一句炸弹拖死。
|
||
"""
|
||
body = (text or "").strip()
|
||
if not body:
|
||
return False
|
||
if _MEASURE_SUPPLEMENT.search(body) and not _FRESH_INQUIRY.search(body):
|
||
return False
|
||
return bool(_FRESH_INQUIRY.search(body))
|
||
|
||
|
||
def scrub_blocked_scan_fields(facts: dict[str, Any]) -> None:
|
||
"""新开一票时，把品名/起运/备注里仍带着违禁词的旧值清掉，避免 field_hit 再拦。"""
|
||
for key in _SCAN_KEYS:
|
||
if contains_blocked(str((facts or {}).get(key) or "")):
|
||
facts[key] = ""
|
||
|
||
|
||
def contains_blocked(text: str) -> bool:
|
||
"""
|
||
这句话里有没有要拦的词。
|
||
|
||
放行说法内部的字不算。例如「玩具枪」放行，「玩具枪支」仍因「枪支」拦下。
|
||
不做英文、拼音、谐音。纯字符串查找，可在 Worker 线程同步调用。
|
||
"""
|
||
raw = text or ""
|
||
if not raw:
|
||
return False
|
||
allow_spans: list[tuple[int, int]] = []
|
||
for phrase in _ALLOW:
|
||
start = 0
|
||
while True:
|
||
found = raw.find(phrase, start)
|
||
if found < 0:
|
||
break
|
||
allow_spans.append((found, found + len(phrase)))
|
||
start = found + len(phrase)
|
||
|
||
def _inside(begin: int, end: int) -> bool:
|
||
return any(left <= begin and end <= right for left, right in allow_spans)
|
||
|
||
for word in _BLOCK:
|
||
start = 0
|
||
while True:
|
||
found = raw.find(word, start)
|
||
if found < 0:
|
||
break
|
||
if not _inside(found, found + len(word)):
|
||
return True
|
||
start = found + len(word)
|
||
return False
|
||
|
||
|
||
def spoken_cargo(text: str) -> str:
|
||
"""
|
||
从这一句里取出销售刚说的品名。
|
||
|
||
「运送衣服」「改成衣服」「换成衣服」「炸弹换衣服」都算。
|
||
取最后一个说法，所以「炸弹换衣服」得到衣服。
|
||
说出来的还是违禁词，或者只是「换一下」，就当没改口。
|
||
"""
|
||
found = ""
|
||
for match in _SPOKEN_CARGO.finditer(text or ""):
|
||
raw = match.group(1) or match.group(2) or ""
|
||
value = _CARGO_TAIL.sub("", raw).strip("，,。；;、 ")
|
||
if not value or value in _CARGO_SKIP or contains_blocked(value):
|
||
continue
|
||
found = value
|
||
return found
|
||
|
||
|
||
def apply_field_edits(text: str, facts: dict[str, Any]) -> dict[str, str]:
|
||
"""
|
||
把当前这句话里的「品名改为…」「起运地改为…」「备注改为…」写进字段。
|
||
|
||
只改这一句写明的项。返回实际写下的键值，方便多段把品名写回每一段。
|
||
会改传入的 facts。值末尾的标点去掉，避免「服装，」留在品名里。
|
||
"""
|
||
written: dict[str, str] = {}
|
||
pairs = list(_FIELD_EDIT.findall(text or ""))
|
||
if not _MULTI_SHEET.search(text or ""):
|
||
pairs.extend(_LABEL_EDIT.findall(text or ""))
|
||
for label, raw_value in pairs:
|
||
value = str(raw_value or "").strip("，,。；;、 ")
|
||
if not value:
|
||
continue
|
||
for key in _WRITE_KEYS.get(label, (label,)):
|
||
facts[key] = value
|
||
written[key] = value
|
||
cargo = spoken_cargo(text)
|
||
if cargo:
|
||
for key in ("品名", "货物品名"):
|
||
facts[key] = cargo
|
||
written[key] = cargo
|
||
return written
|
||
|
||
|
||
def is_cargo_rename(text: str) -> bool:
|
||
"""当前句有没有把品名重新说成一个新值。没说品名的补重量不算改口。"""
|
||
if spoken_cargo(text):
|
||
return True
|
||
body = text or ""
|
||
labels = [label for label, _value in _FIELD_EDIT.findall(body)]
|
||
if not _MULTI_SHEET.search(body):
|
||
labels.extend(label for label, _value in _LABEL_EDIT.findall(body))
|
||
return any(label in {"品名", "货物品名"} for label in labels)
|
||
|
||
|
||
def judge(*, text: str, facts: dict[str, Any], hold: str) -> tuple[bool, str]:
|
||
"""
|
||
决定这一轮私聊能不能继续往建单走。
|
||
|
||
返回 (是否拦截, 新的未改口原话, 本句写下的字段)。
|
||
当前句仍只是在说违禁词：拦住，并把这句记下来。
|
||
当前句已经把品名说成普通货（运送衣服、改成衣服、换成衣服）：前面的违禁原话作废。
|
||
同一句里先提到旧词再改口，按改口后的品名放行。
|
||
只补重量、件数：旧原话继续有效。
|
||
本句无违禁词且像另开一票（查一下…费用）：旧 hold 作废，并清掉字段里残留的违禁词。
|
||
品名改干净之后，起运地或备注里还留着违禁词，仍然拦截。
|
||
会就地更新 facts 里本句写明的品名、起运地、备注。
|
||
第三个返回值是本句写下的键，多段用来写回每一段，避免把别的字段盖掉。
|
||
"""
|
||
edits = apply_field_edits(text, facts)
|
||
replaced = bool(spoken_cargo(text)) or is_cargo_rename(text)
|
||
message_hit = contains_blocked(text)
|
||
fresh = False
|
||
if replaced and not contains_blocked(str(facts.get("品名") or "")):
|
||
# 已经说出新的普通货品名。旧词留在这句里只是在改口，不再记住。
|
||
new_hold = ""
|
||
message_hit = False
|
||
elif message_hit:
|
||
new_hold = (text or "").strip() or (hold or "")
|
||
elif hold and looks_like_fresh_inquiry(text):
|
||
# 本句没有违禁词，且是另开一票（查一下…费用），上一句违禁原话作废。
|
||
# 只补重量/件数不走这里，旧 hold 继续拦。
|
||
fresh = True
|
||
new_hold = ""
|
||
scrub_blocked_scan_fields(facts)
|
||
else:
|
||
new_hold = hold or ""
|
||
field_text = "\n".join(str((facts or {}).get(key) or "") for key in _SCAN_KEYS)
|
||
hold_hit = contains_blocked(new_hold)
|
||
field_hit = contains_blocked(field_text)
|
||
blocked = message_hit or hold_hit or field_hit
|
||
return blocked, new_hold, edits
|