mirror of
https://github.com/zhenxun-org/zhenxun_bot.git
synced 2026-10-05 03:39:59 +08:00
refactor: extract is_ban_cached method to BanConsole
This commit is contained in:
HibiKier
@@ -1,187 +0,0 @@
|
||||
from collections import OrderedDict
|
||||
import random
|
||||
from typing import Any
|
||||
|
||||
from nonebot import on_message
|
||||
from nonebot.adapters import Event
|
||||
from nonebot.plugin import PluginMetadata
|
||||
from nonebot_plugin_alconna import Image, UniMsg
|
||||
from nonebot_plugin_uninfo import Uninfo
|
||||
|
||||
from zhenxun.configs.config import Config
|
||||
from zhenxun.configs.path_config import TEMP_PATH
|
||||
from zhenxun.configs.utils import PluginExtraData, RegisterConfig, Task
|
||||
from zhenxun.utils.common_utils import CommonUtils
|
||||
from zhenxun.utils.enum import PluginType
|
||||
from zhenxun.utils.image_utils import get_download_image_hash
|
||||
from zhenxun.utils.message import MessageUtils
|
||||
|
||||
__plugin_meta__ = PluginMetadata(
|
||||
name="复读",
|
||||
description="群友的本质是什么？是复读机哒！",
|
||||
usage="""
|
||||
usage：
|
||||
重复3次相同的消息时会复读
|
||||
""".strip(),
|
||||
extra=PluginExtraData(
|
||||
author="HibiKier",
|
||||
version="0.2",
|
||||
menu_type="其他",
|
||||
plugin_type=PluginType.DEPENDANT,
|
||||
tasks=[Task(module="fudu", name="复读")],
|
||||
ignore_prompt=True,
|
||||
configs=[
|
||||
RegisterConfig(
|
||||
key="FUDU_PROBABILITY",
|
||||
value=0.7,
|
||||
help="复读概率",
|
||||
default_value=0.7,
|
||||
type=float,
|
||||
),
|
||||
RegisterConfig(
|
||||
module="_task",
|
||||
key="DEFAULT_FUDU",
|
||||
value=True,
|
||||
help="被动 复读 进群默认开关状态",
|
||||
default_value=True,
|
||||
type=bool,
|
||||
),
|
||||
],
|
||||
).to_dict(),
|
||||
)
|
||||
|
||||
|
||||
class Fudu:
|
||||
"""复读数据管理器，使用 LRU 策略限制内存占用"""
|
||||
|
||||
MAX_GROUPS = 500 # 最大缓存群组数量
|
||||
|
||||
def __init__(self):
|
||||
# 使用 OrderedDict 实现 LRU 缓存
|
||||
self._data: OrderedDict[str, dict[str, Any]] = OrderedDict()
|
||||
|
||||
def _get_or_create(self, key: str) -> dict[str, Any]:
|
||||
"""获取或创建群组数据，同时维护 LRU 顺序"""
|
||||
if key in self._data:
|
||||
# 移动到末尾（最近使用）
|
||||
self._data.move_to_end(key)
|
||||
return self._data[key]
|
||||
|
||||
# 如果超出最大限制，删除最旧的条目
|
||||
while len(self._data) >= self.MAX_GROUPS:
|
||||
self._data.popitem(last=False)
|
||||
|
||||
self._data[key] = {"is_repeater": False, "data": []}
|
||||
return self._data[key]
|
||||
|
||||
def append(self, key: str, content: str) -> None:
|
||||
"""添加消息内容"""
|
||||
self._get_or_create(key)["data"].append(content)
|
||||
|
||||
def clear(self, key: str) -> None:
|
||||
"""清空群组的复读数据"""
|
||||
group_data = self._get_or_create(key)
|
||||
group_data["data"] = []
|
||||
group_data["is_repeater"] = False
|
||||
|
||||
def size(self, key: str) -> int:
|
||||
"""获取当前消息数量"""
|
||||
return len(self._get_or_create(key)["data"])
|
||||
|
||||
def check(self, key: str, content: str) -> bool:
|
||||
"""检查内容是否与第一条消息相同"""
|
||||
data_list = self._get_or_create(key)["data"]
|
||||
return bool(data_list) and data_list[0] == content
|
||||
|
||||
def get_first(self, key: str) -> str | None:
|
||||
"""获取第一条消息内容"""
|
||||
data_list = self._get_or_create(key)["data"]
|
||||
return data_list[0] if data_list else None
|
||||
|
||||
def is_repeater(self, key: str) -> bool:
|
||||
"""检查是否已经复读过"""
|
||||
return self._get_or_create(key)["is_repeater"]
|
||||
|
||||
def set_repeater(self, key: str) -> None:
|
||||
"""标记已复读"""
|
||||
self._get_or_create(key)["is_repeater"] = True
|
||||
|
||||
|
||||
_manager = Fudu()
|
||||
|
||||
base_config = Config.get("fudu")
|
||||
|
||||
|
||||
async def rule(message: UniMsg, session: Uninfo, event: Event) -> bool:
|
||||
"""消息匹配规则：仅匹配群聊中的有效消息"""
|
||||
if not session.group:
|
||||
return False
|
||||
if event.is_tome():
|
||||
return False
|
||||
plain_text = message.extract_plain_text().strip()
|
||||
image_list = [m.url for m in message if isinstance(m, Image) and m.url]
|
||||
if not plain_text and not image_list:
|
||||
return False
|
||||
return not await CommonUtils.task_is_block(
|
||||
session, "fudu", session.group.id if session.group else None
|
||||
)
|
||||
|
||||
|
||||
_matcher = on_message(rule=rule, priority=999)
|
||||
|
||||
|
||||
@_matcher.handle()
|
||||
async def _(message: UniMsg, session: Uninfo):
|
||||
# rule 已经确保 session.group 存在
|
||||
group_id = session.group.id # type: ignore
|
||||
|
||||
plain_text = message.extract_plain_text().strip()
|
||||
image_list = [m.url for m in message if isinstance(m, Image) and m.url]
|
||||
|
||||
# 计算图片哈希（如果有图片）
|
||||
img_hash = ""
|
||||
if image_list:
|
||||
img_hash = await get_download_image_hash(image_list[0], group_id)
|
||||
|
||||
add_msg = f"{plain_text}|-|{img_hash}"
|
||||
|
||||
# 更新复读状态
|
||||
if _manager.size(group_id) == 0 or _manager.check(group_id, add_msg):
|
||||
_manager.append(group_id, add_msg)
|
||||
else:
|
||||
_manager.clear(group_id)
|
||||
_manager.append(group_id, add_msg)
|
||||
|
||||
# 检查是否触发复读
|
||||
if _manager.size(group_id) <= 2:
|
||||
return
|
||||
if _manager.is_repeater(group_id):
|
||||
return
|
||||
if random.random() >= base_config.get("FUDU_PROBABILITY"):
|
||||
return
|
||||
|
||||
# 20% 概率打断施法
|
||||
if random.random() < 0.2:
|
||||
is_interrupt = plain_text.replace("打断", "").strip() == "施法"
|
||||
if plain_text.startswith("打断") and is_interrupt:
|
||||
await MessageUtils.build_message(f"打断{plain_text}").finish()
|
||||
else:
|
||||
await MessageUtils.build_message("打断施法！").finish()
|
||||
|
||||
# 执行复读
|
||||
_manager.set_repeater(group_id)
|
||||
|
||||
if image_list and plain_text:
|
||||
result = MessageUtils.build_message(
|
||||
[plain_text, TEMP_PATH / f"compare_download_{group_id}_img.jpg"]
|
||||
)
|
||||
elif image_list:
|
||||
result = MessageUtils.build_message(
|
||||
TEMP_PATH / f"compare_download_{group_id}_img.jpg"
|
||||
)
|
||||
elif plain_text:
|
||||
result = MessageUtils.build_message(plain_text)
|
||||
else:
|
||||
return
|
||||
|
||||
await result.finish()
|
||||
@@ -1,283 +0,0 @@
|
||||
import base64
|
||||
from collections import OrderedDict, deque
|
||||
from pathlib import Path
|
||||
import random
|
||||
from typing import Any
|
||||
|
||||
from nonebot import on_message
|
||||
from nonebot.adapters import Bot, Event
|
||||
from nonebot.adapters.onebot.v11 import Message as V11Message
|
||||
from nonebot.adapters.onebot.v11 import MessageSegment
|
||||
from nonebot.plugin import PluginMetadata
|
||||
from nonebot.utils import run_sync
|
||||
from nonebot_plugin_alconna import Image as alcImg
|
||||
from nonebot_plugin_alconna import UniMsg
|
||||
from nonebot_plugin_uninfo import Uninfo
|
||||
|
||||
from zhenxun.configs.config import BotConfig, Config
|
||||
from zhenxun.configs.path_config import DATA_PATH, TEMP_PATH
|
||||
from zhenxun.configs.utils import PluginExtraData, RegisterConfig, Task
|
||||
from zhenxun.services.log import logger
|
||||
from zhenxun.utils.common_utils import CommonUtils
|
||||
from zhenxun.utils.enum import PluginType
|
||||
from zhenxun.utils.http_utils import AsyncHttpx
|
||||
from zhenxun.utils.image_utils import get_img_hash
|
||||
from zhenxun.utils.message import MessageUtils
|
||||
|
||||
FUDU_IMAGE_PATH = DATA_PATH / "fudu"
|
||||
FUDU_IMAGE_PATH.mkdir(parents=True, exist_ok=True)
|
||||
FUDU_CACHE_PATH = TEMP_PATH / "fudu"
|
||||
FUDU_CACHE_PATH.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# 支持的图片扩展名
|
||||
IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".gif", ".webp"}
|
||||
|
||||
__plugin_meta__ = PluginMetadata(
|
||||
name="复读",
|
||||
description="群友的本质是什么？是复读机哒！",
|
||||
usage="""
|
||||
usage：
|
||||
重复3次相同的消息时会复读
|
||||
""".strip(),
|
||||
extra=PluginExtraData(
|
||||
author="HibiKier",
|
||||
version="0.3",
|
||||
menu_type="其他",
|
||||
plugin_type=PluginType.DEPENDANT,
|
||||
tasks=[Task(module="fudu", name="复读")],
|
||||
ignore_prompt=True,
|
||||
configs=[
|
||||
RegisterConfig(
|
||||
key="FUDU_PROBABILITY",
|
||||
value=0.8,
|
||||
help="复读概率",
|
||||
default_value=0.8,
|
||||
type=float,
|
||||
),
|
||||
RegisterConfig(
|
||||
module="_task",
|
||||
key="DEFAULT_FUDU",
|
||||
value=True,
|
||||
help="被动 复读 进群默认开关状态",
|
||||
default_value=True,
|
||||
type=bool,
|
||||
),
|
||||
RegisterConfig(
|
||||
key="FUDU_TRIGGER_COUNT",
|
||||
value=3,
|
||||
help="触发复读所需的消息重复次数",
|
||||
default_value=3,
|
||||
type=int,
|
||||
),
|
||||
RegisterConfig(
|
||||
key="FUDU_BREAK_PROBABILITY",
|
||||
value=0.2,
|
||||
help="打断复读的概率（基于复读概率）",
|
||||
default_value=0.2,
|
||||
type=float,
|
||||
),
|
||||
RegisterConfig(
|
||||
key="FUDU_BREAK_TEXTS",
|
||||
value=["打断施法！"],
|
||||
help="用于打断复读时随机发送的文本列表",
|
||||
type=list[str],
|
||||
),
|
||||
RegisterConfig(
|
||||
key="FUDU_BREAK_USE_IMAGE",
|
||||
value=True,
|
||||
help="是否启用图片作为打断复读的内容 (图片存放于 data/fudu/ 目录下)",
|
||||
default_value=True,
|
||||
type=bool,
|
||||
),
|
||||
RegisterConfig(
|
||||
key="FUDU_BREAK_USE_TEXT",
|
||||
value=True,
|
||||
help="是否启用文本作为打断复读的内容",
|
||||
default_value=True,
|
||||
type=bool,
|
||||
),
|
||||
],
|
||||
).to_dict(),
|
||||
)
|
||||
|
||||
|
||||
class Fudu:
|
||||
"""复读数据管理器，使用 LRU 策略限制内存占用"""
|
||||
|
||||
MAX_GROUPS = 500 # 最大缓存群组数量
|
||||
|
||||
def __init__(self):
|
||||
self._data: OrderedDict[str, dict[str, Any]] = OrderedDict()
|
||||
|
||||
def _get_or_create(self, key: str) -> dict[str, Any]:
|
||||
"""获取或创建群组数据，同时维护 LRU 顺序"""
|
||||
if key in self._data:
|
||||
self._data.move_to_end(key)
|
||||
return self._data[key]
|
||||
|
||||
# 超出限制时删除最旧的条目
|
||||
while len(self._data) >= self.MAX_GROUPS:
|
||||
self._data.popitem(last=False)
|
||||
|
||||
trigger_count = Config.get_config("fudu", "FUDU_TRIGGER_COUNT", 3)
|
||||
self._data[key] = {
|
||||
"is_repeater": False,
|
||||
"data": deque(maxlen=trigger_count),
|
||||
"message_obj": None,
|
||||
"reply_info": None,
|
||||
}
|
||||
return self._data[key]
|
||||
|
||||
def append(self, key: str, content: str, msg_obj: Any, reply_info: Any) -> None:
|
||||
"""添加消息内容及原始消息对象"""
|
||||
group_data = self._get_or_create(key)
|
||||
group_data["data"].append(content)
|
||||
group_data["message_obj"] = msg_obj
|
||||
group_data["reply_info"] = reply_info
|
||||
|
||||
def clear(self, key: str) -> None:
|
||||
"""清空群组的复读数据"""
|
||||
group_data = self._get_or_create(key)
|
||||
group_data["data"].clear()
|
||||
group_data["is_repeater"] = False
|
||||
group_data["message_obj"] = None
|
||||
group_data["reply_info"] = None
|
||||
|
||||
def size(self, key: str) -> int:
|
||||
"""获取当前消息数量"""
|
||||
return len(self._get_or_create(key)["data"])
|
||||
|
||||
def check(self, key: str, content: str) -> bool:
|
||||
"""检查内容是否与第一条消息相同"""
|
||||
data_list = self._get_or_create(key)["data"]
|
||||
return bool(data_list) and data_list[0] == content
|
||||
|
||||
def get_repeat_target(self, key: str) -> tuple[Any, Any]:
|
||||
"""获取要复读的消息对象和回复信息"""
|
||||
group_data = self._get_or_create(key)
|
||||
return group_data["message_obj"], group_data["reply_info"]
|
||||
|
||||
def is_repeater(self, key: str) -> bool:
|
||||
"""检查是否已经复读过"""
|
||||
return self._get_or_create(key)["is_repeater"]
|
||||
|
||||
def set_repeater(self, key: str) -> None:
|
||||
"""标记已复读"""
|
||||
self._get_or_create(key)["is_repeater"] = True
|
||||
|
||||
|
||||
_manager = Fudu()
|
||||
base_config = Config.get("fudu")
|
||||
|
||||
|
||||
def get_break_images() -> list[Path]:
|
||||
"""获取打断复读用的图片列表"""
|
||||
if not FUDU_IMAGE_PATH.exists():
|
||||
return []
|
||||
return [
|
||||
f
|
||||
for f in FUDU_IMAGE_PATH.iterdir()
|
||||
if f.is_file() and f.suffix.lower() in IMAGE_EXTENSIONS
|
||||
]
|
||||
|
||||
|
||||
async def send_break_response() -> None:
|
||||
"""发送打断复读的响应"""
|
||||
response_pool: list[str | Path] = []
|
||||
|
||||
if base_config.get("FUDU_BREAK_USE_TEXT"):
|
||||
if break_texts := base_config.get("FUDU_BREAK_TEXTS"):
|
||||
response_pool.extend(break_texts)
|
||||
|
||||
if base_config.get("FUDU_BREAK_USE_IMAGE"):
|
||||
response_pool.extend(get_break_images())
|
||||
|
||||
if not response_pool:
|
||||
response_pool.append("打断施法！")
|
||||
|
||||
response = random.choice(response_pool)
|
||||
|
||||
if isinstance(response, Path):
|
||||
file_data = f"base64://{base64.b64encode(response.read_bytes()).decode()}"
|
||||
await _matcher.finish(
|
||||
MessageSegment("image", {"file": file_data, "sub_type": "1"})
|
||||
)
|
||||
else:
|
||||
await MessageUtils.build_message(response).finish()
|
||||
|
||||
|
||||
async def rule(message: UniMsg, session: Uninfo, event: Event) -> bool:
|
||||
"""消息匹配规则：仅匹配群聊中的有效消息"""
|
||||
if not session.group:
|
||||
return False
|
||||
if event.is_tome():
|
||||
return False
|
||||
plain_text = message.extract_plain_text()
|
||||
image_list = [m.url for m in message if isinstance(m, alcImg) and m.url]
|
||||
if not plain_text and not image_list:
|
||||
return False
|
||||
return not await CommonUtils.task_is_block(session, "fudu")
|
||||
|
||||
|
||||
_matcher = on_message(rule=rule, priority=999)
|
||||
|
||||
|
||||
@_matcher.handle()
|
||||
async def _(bot: Bot, message: UniMsg, event: Event, session: Uninfo):
|
||||
# rule 已确保 session.group 存在
|
||||
group_id = session.group.id # type: ignore
|
||||
|
||||
raw_message = event.get_message()
|
||||
reply_info = getattr(event, "reply", None)
|
||||
plain_text = message.extract_plain_text()
|
||||
image_list = [m.url for m in message if isinstance(m, alcImg) and m.url]
|
||||
|
||||
# 检测虚空艾特
|
||||
if plain_text and plain_text.startswith(f"@可爱的{BotConfig.self_nickname}"):
|
||||
await MessageUtils.build_message("复制粘贴的虚空艾特？").send(reply_to=True)
|
||||
|
||||
# 计算图片哈希
|
||||
img_hash = ""
|
||||
if image_list:
|
||||
temp_image_path = FUDU_CACHE_PATH / f"fudu_cache_{group_id}.jpg"
|
||||
try:
|
||||
if await AsyncHttpx.download_file(image_list[0], temp_image_path):
|
||||
img_hash = await run_sync(get_img_hash)(temp_image_path)
|
||||
except Exception as e:
|
||||
logger.warning("下载复读图片以获取Hash时出错", "复读", e=e)
|
||||
|
||||
add_msg = f"{plain_text}|-|{img_hash}"
|
||||
|
||||
# 更新复读状态
|
||||
if _manager.size(group_id) == 0 or _manager.check(group_id, add_msg):
|
||||
_manager.append(group_id, add_msg, raw_message, reply_info)
|
||||
else:
|
||||
_manager.clear(group_id)
|
||||
_manager.append(group_id, add_msg, raw_message, reply_info)
|
||||
|
||||
# 检查是否触发复读
|
||||
trigger_count = base_config.get("FUDU_TRIGGER_COUNT")
|
||||
if _manager.size(group_id) < trigger_count:
|
||||
return
|
||||
if _manager.is_repeater(group_id):
|
||||
return
|
||||
if random.random() >= base_config.get("FUDU_PROBABILITY"):
|
||||
return
|
||||
|
||||
# 判断是否打断复读
|
||||
if random.random() < base_config.get("FUDU_BREAK_PROBABILITY"):
|
||||
_manager.clear(group_id)
|
||||
await send_break_response()
|
||||
|
||||
# 执行复读
|
||||
message_to_send, reply = _manager.get_repeat_target(group_id)
|
||||
if not message_to_send:
|
||||
return
|
||||
|
||||
_manager.set_repeater(group_id)
|
||||
|
||||
# 处理回复消息
|
||||
if reply and isinstance(message_to_send, V11Message):
|
||||
message_to_send.insert(0, MessageSegment.reply(reply.message_id))
|
||||
|
||||
await _matcher.finish(message_to_send)
|
||||
@@ -0,0 +1,20 @@
|
||||
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()))
|
||||
@@ -0,0 +1,122 @@
|
||||
import time
|
||||
|
||||
from pydantic import BaseModel
|
||||
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 = {}
|
||||
"""消息存储"""
|
||||
|
||||
|
||||
class MuteManager:
|
||||
file = DATA_PATH / "group_mute_data.json"
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._group_data: dict[str, GroupData] = {}
|
||||
if self.file.exists():
|
||||
_data = json.load(open(self.file))
|
||||
for gid in _data:
|
||||
self._group_data[gid] = GroupData(
|
||||
count=_data[gid]["count"],
|
||||
time=_data[gid]["time"],
|
||||
duration=_data[gid]["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 = {}
|
||||
for gid in self._group_data:
|
||||
data[gid] = {
|
||||
"count": self._group_data[gid].count,
|
||||
"time": self._group_data[gid].time,
|
||||
"duration": self._group_data[gid].duration,
|
||||
}
|
||||
with open(self.file, "w") 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: 禁言时长
|
||||
"""
|
||||
if group_id not in self._group_data:
|
||||
self._group_data[group_id] = GroupData(
|
||||
count=base_config.get("MUTE_DEFAULT_COUNT"),
|
||||
time=base_config.get("MUTE_DEFAULT_TIME"),
|
||||
duration=base_config.get("MUTE_DEFAULT_DURATION"),
|
||||
)
|
||||
group_data = self._group_data[group_id]
|
||||
if group_data.duration == 0:
|
||||
return 0
|
||||
message_data = group_data.message_data
|
||||
if not message_data.get(user_id):
|
||||
message_data[user_id] = {
|
||||
"time": time.time(),
|
||||
"count": 1,
|
||||
"message": message,
|
||||
}
|
||||
else:
|
||||
if message.find(message_data[user_id]["message"]) != -1:
|
||||
message_data[user_id]["count"] += 1
|
||||
else:
|
||||
message_data[user_id]["time"] = time.time()
|
||||
message_data[user_id]["count"] = 1
|
||||
message_data[user_id]["message"] = message
|
||||
if time.time() - message_data[user_id]["time"] > group_data.time:
|
||||
message_data[user_id]["time"] = time.time()
|
||||
message_data[user_id]["count"] = 1
|
||||
if (
|
||||
message_data[user_id]["count"] > group_data.count
|
||||
and time.time() - message_data[user_id]["time"] < group_data.time
|
||||
):
|
||||
return group_data.duration
|
||||
return 0
|
||||
|
||||
|
||||
mute_manager = MuteManager()
|
||||
@@ -0,0 +1,78 @@
|
||||
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.data_access import DataAccess
|
||||
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
|
||||
ban_dao = DataAccess(BanConsole)
|
||||
if not await ban_dao.safe_get_or_none(
|
||||
user_id=entity_ids.user_id, group_id=entity_ids.group_id
|
||||
):
|
||||
return False
|
||||
if not await ban_dao.safe_get_or_none(user_id="", group_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)
|
||||
@@ -0,0 +1,117 @@
|
||||
from nonebot.plugin import PluginMetadata
|
||||
from nonebot_plugin_alconna import Alconna, Args, Arparma, Match, Option, on_alconna
|
||||
from nonebot_plugin_session import EventSession
|
||||
|
||||
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 ._data_source import base_config, mute_manage
|
||||
|
||||
__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: EventSession,
|
||||
arparma: Arparma,
|
||||
time: Match[int],
|
||||
count: Match[int],
|
||||
duration: Match[int],
|
||||
):
|
||||
group_id = session.id2
|
||||
if not session.id1 or not group_id:
|
||||
return
|
||||
_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_manage.get_group_data(group_id)
|
||||
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_manage.save_data()
|
||||
Reference in New Issue
Block a user