发送与接收消息屏蔽词插件
from functools import partial

from nonebot.log import logger
from nonebot.plugin import PluginMetadata
from nonebot.internal.matcher import Matcher
from nonebot.adapters import Bot, Event, Message
from nonebot.params import CommandArg, CommandStart, EventPlainText

from .hook import send_hook
from .check import blockword_exists
from .config import Config, driver, plugin_config
from .matcher import blockwords_status, blockwords_matcher


@blockwords_matcher.handle()
async def _(
    matcher: Matcher,
    event: Event,
    text: str = EventPlainText(),
):
    if plugin_config.blockwords_user and blockword_exists(text):
        user_id = event.get_user_id()
        logger.warning(f"[{user_id}] 用户触发屏蔽词: {text}")
        if plugin_config.blockwords_stop_propagation:
            matcher.stop_propagation()
    else:
        await matcher.finish()


@blockwords_status.handle()
async def _(matcher: Matcher, msg: Message = CommandArg(), start: str = CommandStart()):
    txt = msg.extract_plain_text().strip()
    if (status := txt == "开启" if txt in ["开启", "关闭"] else None) is not None:
        if start.startswith("用户"):
            plugin_config.blockwords_user = status
            await matcher.finish(f"用户屏蔽词{txt}")
        else:
            plugin_config.blockwords_bot = status
            await matcher.finish(f"机器人屏蔽词{txt}")


@driver.on_bot_connect
async def _(bot: Bot):
    bot.send = partial(send_hook, bot.send)


__plugin_meta__ = PluginMetadata(
    name="屏蔽词插件",
    description="接收的消息或机器人发送的消息进行屏蔽词检查",
    usage="屏蔽词开关：屏蔽词 开启/关闭\n",
    type="application",
    config=Config,
    homepage="https://github.com/MelodyKnit/nonebot-plugin-blockwords",
    supported_adapters=None,
)

nonebot_plugin_blockwords/__init__.py
MelodyKnit/nonebot-plugin-blockwords
import re
from typing import List, Union, Iterable

from jieba import cut

from .utils import get_blockword
from .config import plugin_config

blockwords = get_blockword()
pattern = "|".join(re.escape(w) for w in blockwords)
regex = re.compile(rf"(?:{pattern})")


def replace_text(text: str) -> str:
    if plugin_config.blockwords_replace is None:
        raise ValueError("缺少`blockwords_replace`配置项")
    return plugin_config.blockwords_replace * len(text)


def find_blockword(text: Union[str, Iterable[str]]) -> List[str]:
    """获取屏蔽词文本

    Args:
        text (str): 要检查的文本

    Returns:
        List[str]: 存在的屏蔽词
    """
    if isinstance(text, str):
        if not plugin_config.blockwords_use_jieba:
            return regex.findall(text)
        text = cut(text, cut_all=False)
    return list(set(text) & set(blockwords))


def blockword_exists(text: str) -> bool:
    """是否存在屏蔽词

    Args:
        text (str): 要检查的文本

    Returns:
        bool: 是否存在
    """
    if plugin_config.blockwords_use_jieba:
        return bool(find_blockword(text))
    return bool(regex.search(text))


def blockword_replace(text: str) -> str:
    """屏蔽词替换

    Args:
        text (str): 要检查的文本

    Returns:
        str: 将屏蔽词替换为指定字符后的文本
    """
    if plugin_config.blockwords_replace is None:
        raise ValueError("缺少`blockwords_replace`配置项")
    if plugin_config.blockwords_use_jieba:
        word_segmentation = list(cut(text, cut_all=False))
        if word := find_blockword(word_segmentation):
            for i, w in enumerate(word_segmentation):  # 使用jieba分词后，替换屏蔽词
                if w in word:
                    word_segmentation[i] = replace_text(w)
        return "".join(word_segmentation)
    return regex.sub(lambda m: replace_text(m.group(0)), text)  # type: ignore

nonebot_plugin_blockwords/check.py
MelodyKnit/nonebot-plugin-blockwords
from typing import List, Union, Optional

from nonebot import get_driver
from pydantic import Extra, BaseModel

driver = get_driver()


class Config(BaseModel, extra=Extra.ignore):
    blockwords: List[str] = []  # env中自定义屏蔽词
    blockwords_file: Union[str, List[str]] = []  # 屏蔽词文件路径
    blockwords_priority = 0  # 屏蔽词检查优先级
    blockwords_replace: Optional[str] = None  # 屏蔽词替换字符，不填写时触发屏蔽词时机器人将会不发送消息
    blockwords_use_jieba: bool = True  # 使用jieba分词的方式检查是否触发屏蔽词
    blockwords_bot: bool = True  # 检查机器人发送的消息
    blockwords_user: bool = False  # 检查用户发送的消息
    blockwords_stop_propagation: bool = True  # 用户触发屏蔽词后是否向继续向下传递


plugin_config = Config.parse_obj(driver.config.dict())

nonebot_plugin_blockwords/config.py
MelodyKnit/nonebot-plugin-blockwords
from typing import Any, Type, Union

from nonebot.log import logger
from nonebot.adapters import Event, Message, MessageSegment

from .config import plugin_config
from .check import blockword_exists, blockword_replace


async def send_hook(
    send,
    event: "Event",
    message: Union[str, "Message", "MessageSegment"],
    **kwargs: Any,
):
    if not plugin_config.blockwords_bot:
        return await send(event, message, **kwargs)

    if plugin_config.blockwords_replace is not None:  # 是否带有替换字符
        if isinstance(message, str):
            message = blockword_replace(message)
        elif isinstance(message, Message):
            new_message = message.copy()
            new_message.clear()
            for msg in message:
                if isinstance(msg, MessageSegment) and msg.is_text():
                    new_message += blockword_replace(
                        msg.get_message_class()(msg).extract_plain_text()
                    )
                else:
                    new_message.append(msg)
            message = new_message
        elif isinstance(message, MessageSegment) and message.is_text():
            message_class: Type[Message] = message.get_message_class()
            message = message_class(
                blockword_replace(message_class(message).extract_plain_text())
            )
        await send(event, message, **kwargs)
    else:
        text = message
        if isinstance(message, Message):
            text = message.extract_plain_text()
        elif isinstance(message, MessageSegment) and message.is_text():
            text = message.get_message_class()(message).extract_plain_text()
        if isinstance(text, str) and blockword_exists(text):
            logger.warning(f"屏蔽词触发停止发送消息: {text}")
            return
        await send(event, message, **kwargs)

nonebot_plugin_blockwords/hook.py
MelodyKnit/nonebot-plugin-blockwords
from nonebot.permission import SUPERUSER
from nonebot import on_command, on_message

from .config import plugin_config

blockwords_matcher = on_message(block=False, priority=plugin_config.blockwords_priority)
blockwords_status = on_command(
    "屏蔽词",
    aliases={"用户屏蔽词", "机器人屏蔽词"},
    block=True,
    priority=plugin_config.blockwords_priority,
    permission=SUPERUSER,
)

nonebot_plugin_blockwords/matcher.py
MelodyKnit/nonebot-plugin-blockwords
from typing import List, Callable

from nonebot.params import Depends, EventPlainText

from .check import find_blockword


async def _find_blockword(text: str = EventPlainText()) -> List[str]:
    return find_blockword(text)


FIND_BLOCKWORD: Callable[..., List[str]] = Depends(_find_blockword)

nonebot_plugin_blockwords/params.py
MelodyKnit/nonebot-plugin-blockwords
import json
from typing import List
from pathlib import Path

from nonebot.log import logger

from .config import plugin_config


def get_blockword() -> List[str]:
    """获取屏蔽词

    在配置文件中没有配置blockwords_file时，会读取默认的屏蔽词文件夹中的所有文件

    Returns:
        List[str]: 屏蔽词列表
    """
    words = plugin_config.blockwords.copy()
    if isinstance(plugin_config.blockwords_file, str):
        words.extend(read_words(Path(plugin_config.blockwords_file)))
        logger.success(f"读取屏蔽词文件 << {plugin_config.blockwords_file}")
    elif isinstance(plugin_config.blockwords_file, list):
        for file_path in plugin_config.blockwords_file:
            words.extend(read_words(Path(file_path)))
            logger.success(f"读取屏蔽词文件 << {file_path}")
    words = list(set(words))  # 去重
    words.sort(key=len, reverse=True)  # 按长度排序
    return words


def read_words(file_path: Path) -> List[str]:
    text = file_path.read_text(encoding="utf-8")
    try:
        words = json.loads(text)
        if isinstance(words, list):
            return words
        logger.error(f"{file_path} 屏蔽词文件格式并不是一个 list")
    except ValueError:
        return text.split("\n")
    return []

nonebot_plugin_blockwords/utils.py
MelodyKnit/nonebot-plugin-blockwords