|
|
|
|
|
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 "过滤消息中的敏感词。" |
|
|
