216 lines
8.3 KiB
Python
216 lines
8.3 KiB
Python
from __future__ import annotations
|
||
|
||
from .models import *
|
||
|
||
class IntentResolver:
|
||
"""Resolve raw text to canonical intent."""
|
||
|
||
OPERATION_VERBS = (
|
||
"新增",
|
||
"增加",
|
||
"添加",
|
||
"加",
|
||
"加一个",
|
||
"加个",
|
||
"新建",
|
||
"创建",
|
||
"建立",
|
||
"立项",
|
||
"接入",
|
||
"修复",
|
||
"优化",
|
||
"改成",
|
||
"改为",
|
||
"更新",
|
||
"分配",
|
||
"指派",
|
||
"归档",
|
||
"执行",
|
||
)
|
||
|
||
@staticmethod
|
||
def _looks_like_negated_write_statement(*, text: str) -> bool:
|
||
normalized = re.sub(r"\s+", "", str(text or "")).lower()
|
||
if not normalized:
|
||
return False
|
||
negation_tokens = ("别", "别再", "别乱", "不要", "不要再", "先别", "先不要", "禁止", "不能", "无需", "不用")
|
||
if not any(token in normalized for token in negation_tokens):
|
||
return False
|
||
write_tokens = ("创建", "新建", "新增", "增加", "添加", "加", "接入", "执行", "分配", "指派", "立项")
|
||
object_tokens = ("信号", "需求", "任务", "项目", "事项", "工单", "流程", "信号源")
|
||
if any(w in normalized for w in write_tokens) and any(o in normalized for o in object_tokens):
|
||
return True
|
||
return bool(
|
||
re.search(
|
||
r"(?:别|不要|先别|先不要|禁止|不能).{0,8}(?:创建|新建|新增|增加|添加|加|接入|执行|分配|指派|立项)",
|
||
normalized,
|
||
re.IGNORECASE,
|
||
)
|
||
)
|
||
|
||
@staticmethod
|
||
def resolve(*, text: str, project_id: Optional[int]) -> CanonicalIntent:
|
||
normalized = re.sub(r"\s+", "", str(text or "")).lower()
|
||
pretty = re.sub(r"\s+", " ", str(text or "")).strip()
|
||
if not pretty:
|
||
return CanonicalIntent(INTENT_IGNORE, 1.0, project_id, "", "", {"reason": "empty"})
|
||
if IntentResolver._is_smalltalk_message(pretty=pretty, normalized=normalized):
|
||
return CanonicalIntent(INTENT_IGNORE, 0.95, project_id, "", "", {"reason": "smalltalk"})
|
||
if IntentResolver._looks_like_negated_write_statement(text=pretty):
|
||
return CanonicalIntent(
|
||
INTENT_QUERY,
|
||
0.9,
|
||
project_id,
|
||
"",
|
||
pretty[:300],
|
||
{"reason": "negated_write_statement"},
|
||
)
|
||
explicit_followup_target = FOLLOWUP_WORK_ITEM_ID_RE.search(pretty)
|
||
if explicit_followup_target is not None:
|
||
return CanonicalIntent(
|
||
INTENT_FOLLOWUP_CONTINUE,
|
||
0.92,
|
||
project_id,
|
||
"",
|
||
"",
|
||
{
|
||
"followup_by_work_item_id": True,
|
||
"target_work_item_id": _safe_int(explicit_followup_target.group(1)),
|
||
},
|
||
)
|
||
if any(token in pretty for token in FOLLOWUP_CONFIRM_HINTS):
|
||
return CanonicalIntent(INTENT_FOLLOWUP_CONFIRM, 0.82, project_id, "", "", {})
|
||
if any(token in pretty for token in FOLLOWUP_CONTINUE_HINTS):
|
||
return CanonicalIntent(INTENT_FOLLOWUP_CONTINUE, 0.82, project_id, "", "", {})
|
||
if SignalSourceOrchestrationService.looks_like_signal_source_request(pretty):
|
||
return CanonicalIntent(
|
||
INTENT_SIGNAL_SOURCE_ONBOARD,
|
||
0.92,
|
||
project_id,
|
||
IntentResolver._build_signal_title(pretty, default="新增信号源接入请求"),
|
||
pretty[:300],
|
||
{"source_request": True},
|
||
)
|
||
if (
|
||
IntentResolver._looks_like_project_create_request(text=pretty)
|
||
or extract_project_create_title(pretty)
|
||
):
|
||
return CanonicalIntent(
|
||
INTENT_PROJECT_CREATE,
|
||
0.9,
|
||
project_id,
|
||
IntentResolver._build_signal_title(pretty, default="新项目创建请求"),
|
||
pretty[:300],
|
||
{},
|
||
)
|
||
if IntentResolver._looks_like_signal_create_request(text=pretty):
|
||
return CanonicalIntent(
|
||
INTENT_SIGNAL_CREATE,
|
||
0.78,
|
||
project_id,
|
||
IntentResolver._build_signal_title(pretty, default="新信号"),
|
||
pretty[:300],
|
||
{},
|
||
)
|
||
if IntentResolver._looks_like_query(text=pretty):
|
||
return CanonicalIntent(
|
||
INTENT_QUERY,
|
||
0.88,
|
||
project_id,
|
||
"",
|
||
pretty[:300],
|
||
{},
|
||
)
|
||
return CanonicalIntent(
|
||
INTENT_QUERY,
|
||
0.6,
|
||
project_id,
|
||
"",
|
||
pretty[:300],
|
||
{"fallback": True, "reason": "default_query_fallback"},
|
||
)
|
||
|
||
@staticmethod
|
||
def _looks_like_signal_create_request(*, text: str) -> bool:
|
||
normalized = re.sub(r"\s+", "", str(text or "")).lower()
|
||
if not normalized:
|
||
return False
|
||
if IntentResolver._looks_like_negated_write_statement(text=normalized):
|
||
return False
|
||
candidate_discover_hints = ("候选", "换一批", "刷新", "再来一批", "先分析")
|
||
if (
|
||
any(token in normalized for token in candidate_discover_hints)
|
||
and any(token in normalized for token in ("确认", "入库", "提交"))
|
||
and any(token in normalized for token in ("项目", "project"))
|
||
):
|
||
return True
|
||
if re.search(r"(?:抓|找|给).{0,8}(?:\d+|[一二两三四五])(?:条|个)?候选", normalized, re.IGNORECASE):
|
||
return True
|
||
if any(token in normalized for token in SIGNAL_CREATE_HINTS):
|
||
return True
|
||
signal_object_hints = ("信号", "需求", "任务", "事项", "问题", "改造", "优化")
|
||
has_operation = any(token in normalized for token in IntentResolver.OPERATION_VERBS)
|
||
has_signal_object = any(token in normalized for token in signal_object_hints)
|
||
if has_operation and has_signal_object:
|
||
return True
|
||
return bool(
|
||
re.search(
|
||
r"(?:给|帮|请).{0,20}(?:新增|增加|添加|补充|加).{0,20}(?:信号|需求|任务|事项)",
|
||
normalized,
|
||
re.IGNORECASE,
|
||
)
|
||
)
|
||
|
||
@staticmethod
|
||
def _is_smalltalk_message(*, pretty: str, normalized: str) -> bool:
|
||
compact = re.sub(r"[\\s\\?？!！,，。~～、]+", "", str(pretty or "")).strip().lower()
|
||
if not compact:
|
||
return True
|
||
if any(keyword in compact for keyword in SMALLTALK_STRUCTURED_HINTS):
|
||
return False
|
||
if compact in SMALLTALK_EXACT:
|
||
return True
|
||
if len(compact) <= 8 and any(keyword in compact for keyword in SMALLTALK_SHORT_HINTS):
|
||
return True
|
||
return False
|
||
|
||
@staticmethod
|
||
def _looks_like_query(*, text: str) -> bool:
|
||
normalized = re.sub(r"\s+", "", str(text or "")).lower()
|
||
if not normalized:
|
||
return False
|
||
if IntentResolver._looks_like_signal_create_request(text=normalized):
|
||
return False
|
||
if any(token in normalized for token in QUESTION_HINTS):
|
||
return True
|
||
has_question_structure = (
|
||
"?" in normalized
|
||
or "？" in normalized
|
||
or any(token in normalized for token in ("吗", "呢", "吧", "是否", "是不是"))
|
||
)
|
||
if not has_question_structure:
|
||
return False
|
||
has_operation = any(token in normalized for token in IntentResolver.OPERATION_VERBS)
|
||
return not has_operation
|
||
|
||
@staticmethod
|
||
def _looks_like_project_create_request(*, text: str) -> bool:
|
||
normalized = re.sub(r"\s+", "", str(text or "")).lower()
|
||
if not normalized:
|
||
return False
|
||
if any(token in normalized for token in PROJECT_CREATE_HINTS):
|
||
return True
|
||
return bool(
|
||
re.search(
|
||
r"(?:新增|增加|添加|新建|创建|建立|开个|开一个|建个|建一个|做个|做一个|加个|加一个|立项)(?:一个|个)?项目",
|
||
normalized,
|
||
re.IGNORECASE,
|
||
)
|
||
)
|
||
|
||
@staticmethod
|
||
def _build_signal_title(text: str, *, default: str) -> str:
|
||
raw = str(text or "").strip()
|
||
return raw or default
|
||
|