refactor: enhance UserConsole UID management and remove mute plugin

zhenxun/plugins/mute/__init__.py

@@ -1,20 +0,0 @@
-from pathlib import Path
-
-import nonebot
-from nonebot.plugin import PluginMetadata
-
-from zhenxun.configs.utils import PluginExtraData
-from zhenxun.utils.enum import PluginType
-
-__plugin_meta__ = PluginMetadata(
-    name="刷屏禁言检测",
-    description="",
-    usage="",
-    extra=PluginExtraData(
-        author="HibiKier",
-        version="0.1-473ecd8",
-        plugin_type=PluginType.PARENT,
-    ).to_dict(),
-)
-
-nonebot.load_plugins(str(Path(__file__).parent.resolve()))

zhenxun/plugins/mute/_data_source.py

@@ -1,129 +0,0 @@
-import time
-
-from pydantic import BaseModel, Field
-import ujson as json
-
-from zhenxun.configs.config import Config
-from zhenxun.configs.path_config import DATA_PATH
-
-base_config = Config.get("mute_setting")
-
-
-class GroupData(BaseModel):
-    count: int
-    """次数"""
-    time: int
-    """检测时长"""
-    duration: int
-    """禁言时长"""
-    message_data: dict = Field(default_factory=dict)
-    """消息存储"""
-
-
-class MuteManager:
-    file = DATA_PATH / "group_mute_data.json"
-
-    def __init__(self) -> None:
-        self._group_data: dict[str, GroupData] = {}
-        if self.file.exists():
-            with open(self.file, encoding="utf-8") as f:
-                _data = json.load(f)
-            for gid, gdata in _data.items():
-                self._group_data[gid] = GroupData(
-                    count=gdata["count"],
-                    time=gdata["time"],
-                    duration=gdata["duration"],
-                )
-
-    def get_group_data(self, group_id: str) -> GroupData:
-        """获取群组数据
-
-        参数:
-            group_id: 群组id
-
-        返回:
-            GroupData: GroupData
-        """
-        if group_id not in self._group_data:
-            self._group_data[group_id] = GroupData(
-                count=base_config.get("MUTE_DEFAULT_COUNT", 10) or 10,
-                time=base_config.get("MUTE_DEFAULT_TIME", 7) or 7,
-                duration=base_config.get("MUTE_DEFAULT_DURATION", 10) or 10,
-            )
-        return self._group_data[group_id]
-
-    def reset(self, user_id: str, group_id: str):
-        """重置用户检查次数
-
-        参数:
-            user_id: 用户id
-            group_id: 群组id
-        """
-        if group_data := self._group_data.get(group_id):
-            if user_id in group_data.message_data:
-                group_data.message_data[user_id]["count"] = 0
-
-    def save_data(self):
-        """保存数据"""
-        data = {
-            gid: {
-                "count": gdata.count,
-                "time": gdata.time,
-                "duration": gdata.duration,
-            }
-            for gid, gdata in self._group_data.items()
-        }
-        with open(self.file, "w", encoding="utf-8") as f:
-            json.dump(data, f, indent=4, ensure_ascii=False)
-
-    def add_message(self, user_id: str, group_id: str, message: str) -> int:
-        """添加消息
-
-        参数:
-            user_id: 用户id
-            group_id: 群组id
-            message: 消息内容
-
-        返回:
-            int: 禁言时长
-        """
-        group_data = self.get_group_data(group_id)
-        if group_data.duration == 0:
-            return 0
-
-        message_data = group_data.message_data
-        user_data = message_data.get(user_id)
-        now = time.time()
-
-        if not user_data:
-            message_data[user_id] = {
-                "time": now,
-                "count": 1,
-                "message": message,
-            }
-            return 0
-
-        # 超过检测时间窗口，重置计数
-        if now - user_data["time"] > group_data.time:
-            user_data["time"] = now
-            user_data["count"] = 1
-            user_data["message"] = message
-            return 0
-
-        # 消息内容相似（包含之前的消息），累加计数
-        if user_data["message"] in message:
-            user_data["count"] += 1
-        else:
-            user_data["time"] = now
-            user_data["count"] = 1
-
-        user_data["message"] = message
-
-        # 检测是否触发刷屏
-        if user_data["count"] > group_data.count:
-            return group_data.duration
-
-        return 0
-
-
-mute_manager = MuteManager()

zhenxun/plugins/mute/mute_message.py

@@ -1,74 +0,0 @@
-from nonebot import on_message
-from nonebot.adapters import Bot
-from nonebot.plugin import PluginMetadata
-from nonebot_plugin_alconna import Image, UniMsg
-from nonebot_plugin_uninfo import Uninfo
-
-from zhenxun.configs.config import BotConfig
-from zhenxun.configs.utils import PluginExtraData
-from zhenxun.models.ban_console import BanConsole
-from zhenxun.services.log import logger
-from zhenxun.utils.enum import PluginType
-from zhenxun.utils.image_utils import get_download_image_hash
-from zhenxun.utils.message import MessageUtils
-from zhenxun.utils.platform import PlatformUtils
-from zhenxun.utils.utils import FreqLimiter, get_entity_ids
-
-from ._data_source import mute_manager
-
-__plugin_meta__ = PluginMetadata(
-    name="刷屏监听",
-    description="这是刷屏检测的监听器，用于检测用户是否在规定时间内发送了相同的消息",
-    usage="无",
-    extra=PluginExtraData(
-        author="HibiKier",
-        version="0.1-473ecd8",
-        menu_type="其他",
-        plugin_type=PluginType.DEPENDANT,
-    ).to_dict(),
-)
-
-
-async def rule(session: Uninfo) -> bool:
-    entity_ids = get_entity_ids(session)
-    if not session.group:
-        return False
-    if mute_manager.get_group_data(entity_ids.group_id or "0").duration == 0:
-        return False
-    if await BanConsole.is_ban_cached(entity_ids.user_id, entity_ids.group_id):
-        return False
-    return True
-
-
-_matcher = on_message(rule=rule, priority=1, block=False)
-
-_flmt = FreqLimiter(30)
-
-
-@_matcher.handle()
-async def _(bot: Bot, session: Uninfo, message: UniMsg):
-    entity_ids = get_entity_ids(session)
-    plain_text = message.extract_plain_text()
-    image_list = [m.url for m in message if isinstance(m, Image) and m.url]
-    img_hash = ""
-    for url in image_list:
-        img_hash += await get_download_image_hash(url, "_mute_")
-    _message = plain_text + img_hash
-    if duration := mute_manager.add_message(
-        entity_ids.user_id, entity_ids.group_id or "0", _message
-    ):
-        try:
-            if _flmt.check(entity_ids.user_id):
-                _flmt.start_cd(entity_ids.user_id)
-                await PlatformUtils.ban_user(
-                    bot, entity_ids.user_id, entity_ids.group_id or "0", duration
-                )
-                await MessageUtils.build_message(
-                    f"检测到恶意刷屏，{BotConfig.self_nickname}要把你关进小黑屋！"
-                ).send(at_sender=True)
-                mute_manager.reset(entity_ids.user_id, entity_ids.group_id or "0")
-                logger.info(
-                    f"检测刷屏 被禁言 {duration} 分钟", "禁言检查", session=session
-                )
-        except Exception as e:
-            logger.error("禁言发送错误", "禁言检测", session=session, e=e)

zhenxun/plugins/mute/mute_setting.py

@@ -1,116 +0,0 @@
-from nonebot.plugin import PluginMetadata
-from nonebot_plugin_alconna import Alconna, Args, Arparma, Match, Option, on_alconna
-from nonebot_plugin_uninfo import Uninfo
-
-from zhenxun.configs.config import BotConfig
-from zhenxun.configs.utils import PluginExtraData, RegisterConfig
-from zhenxun.services.log import logger
-from zhenxun.utils.enum import PluginType
-from zhenxun.utils.message import MessageUtils
-from zhenxun.utils.rules import ensure_group
-from zhenxun.utils.utils import get_entity_ids
-
-from ._data_source import base_config, mute_manager
-
-__plugin_meta__ = PluginMetadata(
-    name="刷屏禁言",
-    description="刷屏禁言相关操作",
-    usage=f"""
-    刷屏禁言相关操作，需要 {BotConfig.self_nickname} 有群管理员权限
-    指令：
-        刷屏设置: 查看当前设置
-        -c [count]: 检测最大次数
-        -t [time]: 规定时间内
-        -d [duration]: 禁言时长
-        示例:
-            刷屏设置 -c 10: 设置最大次数为10
-            刷屏设置 -t 100 -d 20: 设置规定时间和禁言时长
-            刷屏设置 -d 10: 设置禁言时长为10
-        * 即 X 秒内发送同样消息 N 次，禁言 M 分钟 *
-    """.strip(),
-    extra=PluginExtraData(
-        author="HibiKier",
-        version="0.1-473ecd8",
-        menu_type="其他",
-        plugin_type=PluginType.ADMIN,
-        admin_level=base_config.get("MUTE_LEVEL", 5),
-        configs=[
-            RegisterConfig(
-                key="MUTE_LEVEL",
-                value=5,
-                help="更改禁言设置的管理权限",
-                default_value=5,
-                type=int,
-            ),
-            RegisterConfig(
-                key="MUTE_DEFAULT_COUNT",
-                value=10,
-                help="刷屏禁言默认检测次数",
-                default_value=10,
-                type=int,
-            ),
-            RegisterConfig(
-                key="MUTE_DEFAULT_TIME",
-                value=7,
-                help="刷屏检测默认规定时间",
-                default_value=7,
-                type=int,
-            ),
-            RegisterConfig(
-                key="MUTE_DEFAULT_DURATION",
-                value=10,
-                help="刷屏检测默禁言时长（分钟）",
-                default_value=10,
-                type=int,
-            ),
-        ],
-    ).to_dict(),
-)
-
-
-_setting_matcher = on_alconna(
-    Alconna(
-        "刷屏设置",
-        Option("-t|--time", Args["time", int], help_text="检测时长"),
-        Option("-c|--count", Args["count", int], help_text="检测次数"),
-        Option("-d|--duration", Args["duration", int], help_text="禁言时长"),
-    ),
-    rule=ensure_group,
-    block=True,
-    priority=5,
-)
-
-
-@_setting_matcher.handle()
-async def _(
-    session: Uninfo,
-    arparma: Arparma,
-    time: Match[int],
-    count: Match[int],
-    duration: Match[int],
-):
-    entity_ids = get_entity_ids(session)
-    _time = time.result if time.available else None
-    _count = count.result if count.available else None
-    _duration = duration.result if duration.available else None
-    group_data = mute_manager.get_group_data(entity_ids.group_id or "0")
-    if _time is None and _count is None and _duration is None:
-        await MessageUtils.build_message(
-            f"最大次数：{group_data.count} 次\n"
-            f"规定时间：{group_data.time} 秒\n"
-            f"禁言时长：{group_data.duration:.2f} 分钟\n"
-            f"【在规定时间内发送相同消息超过最大次数则禁言\n当禁言时长为0时关闭此功能】"
-        ).finish(reply_to=True)
-    if _time is not None:
-        group_data.time = _time
-    if _count is not None:
-        group_data.count = _count
-    if _duration is not None:
-        group_data.duration = _duration
-    await MessageUtils.build_message("设置成功!").send(reply_to=True)
-    logger.info(
-        f"设置禁言配置 time: {_time}, count: {_count}, duration: {_duration}",
-        arparma.header_result,
-        session=session,
-    )
-    mute_manager.save_data()