| |
|
|
| import json |
| import os |
|
|
| import plugins |
| from bridge.context import ContextType |
| from bridge.reply import Reply, ReplyType |
| from common.log import logger |
| from plugins import * |
|
|
| from .lib.WordsSearch import WordsSearch |
|
|
|
|
| @plugins.register( |
| name="Banwords", |
| desire_priority=100, |
| hidden=True, |
| desc="判断消息中是否有敏感词、决定是否回复。", |
| version="1.0", |
| author="lanvent", |
| ) |
| class Banwords(Plugin): |
| def __init__(self): |
| super().__init__() |
| try: |
| |
| conf = super().load_config() |
| curdir = os.path.dirname(__file__) |
| if not conf: |
| |
| config_path = os.path.join(curdir, "config.json") |
| if not os.path.exists(config_path): |
| conf = {"action": "ignore"} |
| with open(config_path, "w") as f: |
| json.dump(conf, f, indent=4) |
|
|
| self.searchr = WordsSearch() |
| self.action = conf["action"] |
| banwords_path = os.path.join(curdir, "banwords.txt") |
| with open(banwords_path, "r", encoding="utf-8") as f: |
| words = [] |
| for line in f: |
| word = line.strip() |
| if word: |
| words.append(word) |
| self.searchr.SetKeywords(words) |
| self.handlers[Event.ON_HANDLE_CONTEXT] = self.on_handle_context |
| if conf.get("reply_filter", True): |
| self.handlers[Event.ON_DECORATE_REPLY] = self.on_decorate_reply |
| self.reply_action = conf.get("reply_action", "ignore") |
| logger.info("[Banwords] inited") |
| except Exception as e: |
| logger.warn("[Banwords] init failed, ignore or see https://github.com/zhayujie/chatgpt-on-wechat/tree/master/plugins/banwords .") |
| raise e |
|
|
| def on_handle_context(self, e_context: EventContext): |
| if e_context["context"].type not in [ |
| ContextType.TEXT, |
| ContextType.IMAGE_CREATE, |
| ]: |
| return |
|
|
| content = e_context["context"].content |
| logger.debug("[Banwords] on_handle_context. content: %s" % content) |
| if self.action == "ignore": |
| f = self.searchr.FindFirst(content) |
| if f: |
| logger.info("[Banwords] %s in message" % f["Keyword"]) |
| e_context.action = EventAction.BREAK_PASS |
| return |
| elif self.action == "replace": |
| if self.searchr.ContainsAny(content): |
| reply = Reply(ReplyType.INFO, "发言中包含敏感词，请重试: \n" + self.searchr.Replace(content)) |
| e_context["reply"] = reply |
| e_context.action = EventAction.BREAK_PASS |
| return |
|
|
| def on_decorate_reply(self, e_context: EventContext): |
| if e_context["reply"].type not in [ReplyType.TEXT]: |
| return |
|
|
| reply = e_context["reply"] |
| content = reply.content |
| if self.reply_action == "ignore": |
| f = self.searchr.FindFirst(content) |
| if f: |
| logger.info("[Banwords] %s in reply" % f["Keyword"]) |
| e_context["reply"] = None |
| e_context.action = EventAction.BREAK_PASS |
| return |
| elif self.reply_action == "replace": |
| if self.searchr.ContainsAny(content): |
| reply = Reply(ReplyType.INFO, "已替换回复中的敏感词: \n" + self.searchr.Replace(content)) |
| e_context["reply"] = reply |
| e_context.action = EventAction.CONTINUE |
| return |
|
|
| def get_help_text(self, **kwargs): |
| return "过滤消息中的敏感词。" |
|
|
