Parcourir la source

fix: make member shortcuts reliably dispatch

AI-Co-Authored-By: Codex
chendeben il y a 1 semaine
Parent
commit
b834782d3e
4 fichiers modifiés avec 110 ajouts et 23 suppressions
  1. 3 1
      tests/conftest.py
  2. 63 0
      tests/test_admin_shortcuts.py
  3. 37 16
      wbb/modules/admin.py
  4. 7 6
      wbb/services/member_shortcuts.py

+ 3 - 1
tests/conftest.py

@@ -8,6 +8,7 @@ from types import SimpleNamespace
 
 import pytest
 from mongomock_motor import AsyncMongoMockClient
+from pyrogram import filters
 from pyrogram.enums import ChatMemberStatus
 
 ROOT = Path(__file__).resolve().parents[1]
@@ -111,6 +112,7 @@ def app_modules(tmp_path):
     fake_wbb.db = AsyncMongoMockClient().wbb
     fake_wbb.control_db = fake_wbb.db
     fake_wbb.app = fake_app
+    fake_wbb.aiohttpsession = None
     fake_wbb.log = SimpleNamespace(info=lambda _message: None, error=lambda _message: None)
     fake_wbb.BOT_ID = 999
     fake_wbb.BOT_NAME = "Test Bot"
@@ -122,7 +124,7 @@ def app_modules(tmp_path):
     fake_wbb.SUPERVISOR_MODE = False
     fake_wbb.ADMIN_BOOTSTRAP_MODE = False
     fake_wbb.BOT_SUPERVISOR = None
-    fake_wbb.SUDOERS = set()
+    fake_wbb.SUDOERS = filters.user()
     fake_wbb.MOD_LOAD = []
     fake_wbb.MOD_NOLOAD = []
     fake_wbb.MESSAGE_DUMP_CHAT = 0

+ 63 - 0
tests/test_admin_shortcuts.py

@@ -1,5 +1,9 @@
 from __future__ import annotations
 
+from datetime import datetime
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
 import pytest
 
 
@@ -12,6 +16,9 @@ import pytest
         ("解除禁言", ("/unmute", ["unmute"])),
         ("警告", ("/warn", ["warn"])),
         ("禁言10分钟", ("/tmute 10m", ["tmute", "10m"])),
+        ("  禁言1分钟  ", ("/tmute 1m", ["tmute", "1m"])),
+        ("禁言1分钟", ("/tmute 1m", ["tmute", "1m"])),
+        ("禁言\u200b1分钟", ("/tmute 1m", ["tmute", "1m"])),
         ("禁言 2 小时", ("/tmute 2h", ["tmute", "2h"])),
         ("禁言1天", ("/tmute 1d", ["tmute", "1d"])),
         ("踢出", ("/kick", ["kick"])),
@@ -28,3 +35,59 @@ def test_member_action_shortcut_parser(app_modules, text, expected):
 def test_member_action_shortcut_rejects_ambiguous_text(app_modules, text):
     shortcuts = app_modules.load("wbb.services.member_shortcuts")
     assert shortcuts.member_action_shortcut_command(text) is None
+
+
+@pytest.mark.asyncio
+async def test_member_action_shortcut_executes_temporary_mute(
+    app_modules, monkeypatch
+):
+    admin = app_modules.load("wbb.modules.admin")
+    chat_id = -100123
+    actor_id = 10
+    target_id = 20
+    privileges = SimpleNamespace(
+        can_post_messages=False,
+        can_edit_messages=False,
+        can_delete_messages=True,
+        can_restrict_members=True,
+        can_promote_members=False,
+        can_change_info=False,
+        can_invite_users=False,
+        can_pin_messages=False,
+        can_manage_video_chats=False,
+    )
+    app_modules.app.members[(chat_id, actor_id)] = SimpleNamespace(
+        privileges=privileges
+    )
+    app_modules.app.get_users = AsyncMock(
+        return_value=SimpleNamespace(mention="@target")
+    )
+    monkeypatch.setattr(admin, "list_admins", AsyncMock(return_value=[]))
+
+    restrict_member = AsyncMock()
+    target_reply = AsyncMock()
+    replied_message = SimpleNamespace(
+        from_user=SimpleNamespace(id=target_id),
+        sender_chat=None,
+        reply_text=target_reply,
+    )
+    message = SimpleNamespace(
+        text="  禁言1分钟\u200b  ",
+        command=None,
+        chat=SimpleNamespace(id=chat_id, restrict_member=restrict_member),
+        from_user=SimpleNamespace(id=actor_id, mention="@admin"),
+        sender_chat=None,
+        reply_to_message=replied_message,
+        reply_text=AsyncMock(),
+    )
+
+    assert await admin._member_action_shortcut_filter(None, None, message)
+    await admin.member_action_shortcut(None, message)
+
+    restrict_member.assert_awaited_once()
+    args, kwargs = restrict_member.await_args
+    assert args == (target_id,)
+    assert kwargs["until_date"] > datetime.now()
+    assert message.text == "  禁言1分钟\u200b  "
+    assert message.command is None
+    assert "禁言时长:** 1m" in target_reply.await_args.args[0]

+ 37 - 16
wbb/modules/admin.py

@@ -42,7 +42,6 @@ from wbb.core.decorators.errors import capture_err
 from wbb.core.keyboard import ikb
 from wbb.services.bot_permissions import has_permission
 from wbb.services.member_shortcuts import (
-    MEMBER_ACTION_SHORTCUT_RE,
     member_action_shortcut_command,
 )
 from wbb.utils.dbfunctions import (
@@ -632,29 +631,51 @@ async def mute(_, message: Message):
     await message.reply_text(msg, reply_markup=keyboard)
 
 
+async def _member_action_shortcut_filter(_, __, message: Message) -> bool:
+    return member_action_shortcut_command(message.text or "") is not None
+
+
+member_action_shortcut_filter = filters.create(
+    _member_action_shortcut_filter,
+    "MemberActionShortcutFilter",
+)
+
+
 @app.on_message(
-    filters.group
-    & filters.reply
-    & filters.regex(MEMBER_ACTION_SHORTCUT_RE)
+    filters.group & filters.reply & member_action_shortcut_filter,
+    group=-999,
 )
 async def member_action_shortcut(client, message: Message):
     if not has_permission("chat.members", BOT_PERMISSIONS):
-        return None
+        return await message.reply_text(
+            "当前机器人角色未启用成员管理权限。"
+        )
     command = member_action_shortcut_command(message.text or "")
     if not command or not message.reply_to_message:
         return None
+    original_text = message.text
+    original_command = message.command
     message.text, message.command = command
-    if message.command[0] == "ban":
-        return await banFunc(client, message)
-    if message.command[0] in {"mute", "tmute"}:
-        return await mute(client, message)
-    if message.command[0] == "warn":
-        return await warn_user(client, message)
-    if message.command[0] == "unban":
-        return await unban_func(client, message)
-    if message.command[0] == "unmute":
-        return await unmute(client, message)
-    return await kickFunc(client, message)
+    log.info(
+        f"成员快捷操作:chat={message.chat.id} "
+        f"actor={getattr(message.from_user, 'id', 0)} "
+        f"action={message.command[0]}"
+    )
+    try:
+        if message.command[0] == "ban":
+            return await banFunc(client, message)
+        if message.command[0] in {"mute", "tmute"}:
+            return await mute(client, message)
+        if message.command[0] == "warn":
+            return await warn_user(client, message)
+        if message.command[0] == "unban":
+            return await unban_func(client, message)
+        if message.command[0] == "unmute":
+            return await unmute(client, message)
+        return await kickFunc(client, message)
+    finally:
+        message.text = original_text
+        message.command = original_command
 
 
 # Unmute members

+ 7 - 6
wbb/services/member_shortcuts.py

@@ -1,15 +1,16 @@
 from __future__ import annotations
 
 import re
-
-MEMBER_ACTION_SHORTCUT_RE = (
-    r"^(?:封禁|解封|解禁|解除封禁|禁言|解除禁言|警告|"
-    r"禁言\s*[1-9]\d?\s*(?:分钟|分|小时|时|天)|踢出|踢|飞)$"
-)
+import unicodedata
 
 
 def member_action_shortcut_command(text: str) -> tuple[str, list[str]] | None:
-    normalized = text.strip()
+    normalized = (
+        unicodedata.normalize("NFKC", text)
+        .replace("\u200b", "")
+        .replace("\ufeff", "")
+        .strip()
+    )
     if normalized == "封禁":
         return "/ban", ["ban"]
     if normalized == "禁言":