102 lines
3.0 KiB
Python
102 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any, Optional
|
|
|
|
import re
|
|
|
|
|
|
def _normalize_token(value: Any) -> str:
|
|
text = re.sub(r"\s+", "", str(value or "").strip())
|
|
return text[:128]
|
|
|
|
|
|
def has_feishu_reply_anchor(thread_context: Optional[dict[str, Any]] = None) -> bool:
|
|
if not isinstance(thread_context, dict):
|
|
return False
|
|
return bool(_normalize_token(thread_context.get("reply_to_message_id")))
|
|
|
|
|
|
def has_feishu_bot_mention(
|
|
*,
|
|
mentioned_bot_names: Optional[list[Any]] = None,
|
|
message_content: Optional[str] = None,
|
|
) -> bool:
|
|
normalized_names = {
|
|
str(item or "").strip().lower()
|
|
for item in (mentioned_bot_names or [])
|
|
if str(item or "").strip()
|
|
}
|
|
if any(name in {"pmflow", "_user_1"} for name in normalized_names):
|
|
return True
|
|
|
|
normalized_text = str(message_content or "").strip().lower()
|
|
if not normalized_text:
|
|
return False
|
|
return "@pmflow" in normalized_text or "@_user_1" in normalized_text
|
|
|
|
|
|
def has_feishu_explicit_group_trigger(
|
|
*,
|
|
mentioned_bot_names: Optional[list[Any]] = None,
|
|
message_content: Optional[str] = None,
|
|
thread_context: Optional[dict[str, Any]] = None,
|
|
) -> bool:
|
|
return has_feishu_bot_mention(
|
|
mentioned_bot_names=mentioned_bot_names,
|
|
message_content=message_content,
|
|
) or has_feishu_reply_anchor(thread_context)
|
|
|
|
|
|
def extract_thread_anchor(thread_context: Optional[dict[str, Any]] = None) -> str:
|
|
if not isinstance(thread_context, dict):
|
|
return ""
|
|
for key in ("root_message_id", "thread_id", "reply_to_message_id"):
|
|
value = _normalize_token(thread_context.get(key))
|
|
if value:
|
|
return value
|
|
return ""
|
|
|
|
|
|
def build_feishu_thread_key(
|
|
*,
|
|
source_chat_id: Optional[str],
|
|
source_sender_open_id: Optional[str],
|
|
sender: Optional[str],
|
|
source_message_id: Optional[str] = None,
|
|
thread_context: Optional[dict[str, Any]] = None,
|
|
) -> str:
|
|
chat = _normalize_token(source_chat_id)
|
|
anchor = extract_thread_anchor(thread_context)
|
|
if anchor:
|
|
return f"feishu:{chat}:{anchor}" if chat else f"feishu:{anchor}"
|
|
|
|
actor = _normalize_token(source_sender_open_id)
|
|
if not actor:
|
|
actor = _normalize_token(sender).lower()
|
|
if actor:
|
|
return f"feishu:{chat}:{actor}" if chat else f"feishu:{actor}"
|
|
|
|
message_id = _normalize_token(source_message_id)
|
|
if message_id:
|
|
return f"feishu:{chat}:{message_id}" if chat else f"feishu:{message_id}"
|
|
return ""
|
|
|
|
|
|
def build_feishu_conversation_scope_key(
|
|
*,
|
|
source_chat_id: Optional[str],
|
|
source_sender_open_id: Optional[str],
|
|
sender: Optional[str],
|
|
) -> str:
|
|
chat = _normalize_token(source_chat_id)
|
|
actor = _normalize_token(source_sender_open_id)
|
|
if not actor:
|
|
actor = _normalize_token(sender).lower()
|
|
if chat and actor:
|
|
return f"feishu:{chat}:{actor}"
|
|
if actor:
|
|
return f"feishu:{actor}"
|
|
if chat:
|
|
return f"feishu:{chat}"
|
|
return ""
|