zhilv
ee1bd583d8
- 新增 `#四个中文字+空格` 消息匹配规则，可配置前缀和长度 - 匹配成功后 POST 到 COMMAND_CALLBACK_URL，携带命令名、内容、用户信息 - 使用 EventMixin.events() 订阅消息流，on_close 自动取消监听 - 新增配置项：COMMAND_PREFIX、COMMAND_LENGTH、COMMAND_CALLBACK_URL - 更新 .env.example 和 README 文档
58 lines
1.9 KiB
Python
58 lines
1.9 KiB
Python
"""命令监听处理器：匹配 #命令名+空格 格式的消息，转发到外部回调 URL。"""
|
|
|
|
import re
|
|
|
|
import aiohttp
|
|
|
|
from ..config import COMMAND_CALLBACK_URL, COMMAND_LENGTH, COMMAND_PREFIX
|
|
from ..response import error, ok
|
|
|
|
|
|
def build_command_pattern() -> re.Pattern:
|
|
"""构建命令匹配正则：# + N个中文字 + 空格。"""
|
|
return re.compile(
|
|
rf"^{re.escape(COMMAND_PREFIX)}([\u4e00-\u9fff]{{{COMMAND_LENGTH}}})\s+(.+)",
|
|
re.DOTALL,
|
|
)
|
|
|
|
|
|
COMMAND_PATTERN = build_command_pattern()
|
|
|
|
|
|
def parse_command(raw_message: str) -> dict | None:
|
|
"""解析消息，匹配命令模式。返回 {command, content, raw_message} 或 None。"""
|
|
match = COMMAND_PATTERN.match(raw_message.strip())
|
|
if not match:
|
|
return None
|
|
return {
|
|
"command": match.group(1),
|
|
"content": match.group(2).strip(),
|
|
"raw_message": raw_message.strip(),
|
|
}
|
|
|
|
|
|
async def send_command_callback(data: dict, logger) -> bool:
|
|
"""将命令数据 POST 到外部回调 URL。返回是否成功。"""
|
|
if not COMMAND_CALLBACK_URL:
|
|
logger.warning("COMMAND_CALLBACK_URL 未配置，跳过命令回调")
|
|
return False
|
|
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(
|
|
COMMAND_CALLBACK_URL,
|
|
json=data,
|
|
timeout=aiohttp.ClientTimeout(total=10),
|
|
) as resp:
|
|
if resp.status >= 400:
|
|
body = await resp.text()
|
|
logger.error(
|
|
"命令回调失败: status=%d url=%s body=%s",
|
|
resp.status, COMMAND_CALLBACK_URL, body[:200],
|
|
)
|
|
return False
|
|
return True
|
|
except Exception as exc:
|
|
logger.error("命令回调异常: url=%s error=%s", COMMAND_CALLBACK_URL, exc)
|
|
return False
|