给keywords_reaction增加正则表达式匹配

顺便做了正则表达式预编译
This commit is contained in:
HexatomicRing
2025-04-10 10:29:55 +08:00
parent dbc60dbfee
commit 360406efde
7 changed files with 37 additions and 8 deletions

View File

@@ -2,7 +2,6 @@ from src.common.logger import get_module_logger
from src.plugins.chat.message import MessageRecv from src.plugins.chat.message import MessageRecv
from src.plugins.storage.storage import MessageStorage from src.plugins.storage.storage import MessageStorage
from src.plugins.config.config import global_config from src.plugins.config.config import global_config
import re
from datetime import datetime from datetime import datetime
logger = get_module_logger("pfc_message_processor") logger = get_module_logger("pfc_message_processor")
@@ -28,7 +27,7 @@ class MessageProcessor:
def _check_ban_regex(self, text: str, chat, userinfo) -> bool: def _check_ban_regex(self, text: str, chat, userinfo) -> bool:
"""检查消息是否匹配过滤正则表达式""" """检查消息是否匹配过滤正则表达式"""
for pattern in global_config.ban_msgs_regex: for pattern in global_config.ban_msgs_regex:
if re.search(pattern, text): if pattern.search(text):
logger.info( logger.info(
f"[{chat.group_info.group_name if chat.group_info else '私聊'}]{userinfo.user_nickname}:{text}" f"[{chat.group_info.group_name if chat.group_info else '私聊'}]{userinfo.user_nickname}:{text}"
) )

View File

@@ -1,6 +1,6 @@
import time import time
from random import random from random import random
import re
from typing import List from typing import List
from ...memory_system.Hippocampus import HippocampusManager from ...memory_system.Hippocampus import HippocampusManager
from ...moods.moods import MoodManager from ...moods.moods import MoodManager
@@ -302,7 +302,7 @@ class ReasoningChat:
def _check_ban_regex(self, text: str, chat, userinfo) -> bool: def _check_ban_regex(self, text: str, chat, userinfo) -> bool:
"""检查消息是否匹配过滤正则表达式""" """检查消息是否匹配过滤正则表达式"""
for pattern in global_config.ban_msgs_regex: for pattern in global_config.ban_msgs_regex:
if re.search(pattern, text): if pattern.search(text):
logger.info( logger.info(
f"[{chat.group_info.group_name if chat.group_info else '私聊'}]{userinfo.user_nickname}:{text}" f"[{chat.group_info.group_name if chat.group_info else '私聊'}]{userinfo.user_nickname}:{text}"
) )

View File

@@ -115,6 +115,17 @@ class PromptBuilder:
f"检测到以下关键词之一:{rule.get('keywords', [])},触发反应:{rule.get('reaction', '')}" f"检测到以下关键词之一:{rule.get('keywords', [])},触发反应:{rule.get('reaction', '')}"
) )
keywords_reaction_prompt += rule.get("reaction", "") + "" keywords_reaction_prompt += rule.get("reaction", "") + ""
for pattern in rule.get("regex", []):
result = pattern.search(message_txt)
if result:
reaction = rule.get('reaction', '')
for name, content in result.groupdict().items():
reaction = reaction.replace(f'[{name}]', content)
logger.info(
f"匹配到以下正则表达式:{pattern},触发反应:{reaction}"
)
keywords_reaction_prompt += reaction + ""
break
# 中文高手(新加的好玩功能) # 中文高手(新加的好玩功能)
prompt_ger = "" prompt_ger = ""

View File

@@ -1,6 +1,5 @@
import time import time
from random import random from random import random
import re
import traceback import traceback
from typing import List from typing import List
from ...memory_system.Hippocampus import HippocampusManager from ...memory_system.Hippocampus import HippocampusManager
@@ -388,7 +387,7 @@ class ThinkFlowChat:
def _check_ban_regex(self, text: str, chat, userinfo) -> bool: def _check_ban_regex(self, text: str, chat, userinfo) -> bool:
"""检查消息是否匹配过滤正则表达式""" """检查消息是否匹配过滤正则表达式"""
for pattern in global_config.ban_msgs_regex: for pattern in global_config.ban_msgs_regex:
if re.search(pattern, text): if pattern.search(text):
logger.info( logger.info(
f"[{chat.group_info.group_name if chat.group_info else '私聊'}]{userinfo.user_nickname}:{text}" f"[{chat.group_info.group_name if chat.group_info else '私聊'}]{userinfo.user_nickname}:{text}"
) )

View File

@@ -61,6 +61,17 @@ class PromptBuilder:
f"检测到以下关键词之一:{rule.get('keywords', [])},触发反应:{rule.get('reaction', '')}" f"检测到以下关键词之一:{rule.get('keywords', [])},触发反应:{rule.get('reaction', '')}"
) )
keywords_reaction_prompt += rule.get("reaction", "") + "" keywords_reaction_prompt += rule.get("reaction", "") + ""
for pattern in rule.get("regex", []):
result = pattern.search(message_txt)
if result:
reaction = rule.get('reaction', '')
for name, content in result.groupdict().items():
reaction = reaction.replace(f'[{name}]', content)
logger.info(
f"匹配到以下正则表达式:{pattern},触发反应:{reaction}"
)
keywords_reaction_prompt += reaction + ""
break
# 中文高手(新加的好玩功能) # 中文高手(新加的好玩功能)
prompt_ger = "" prompt_ger = ""

View File

@@ -1,4 +1,5 @@
import os import os
import re
from dataclasses import dataclass, field from dataclasses import dataclass, field
from typing import Dict, List, Optional from typing import Dict, List, Optional
from dateutil import tz from dateutil import tz
@@ -545,8 +546,8 @@ class BotConfig:
"response_interested_rate_amplifier", config.response_interested_rate_amplifier "response_interested_rate_amplifier", config.response_interested_rate_amplifier
) )
config.down_frequency_rate = msg_config.get("down_frequency_rate", config.down_frequency_rate) config.down_frequency_rate = msg_config.get("down_frequency_rate", config.down_frequency_rate)
config.ban_msgs_regex = msg_config.get("ban_msgs_regex", config.ban_msgs_regex) for r in msg_config.get("ban_msgs_regex", config.ban_msgs_regex):
config.ban_msgs_regex.add(re.compile(r))
if config.INNER_VERSION in SpecifierSet(">=0.0.11"): if config.INNER_VERSION in SpecifierSet(">=0.0.11"):
config.max_response_length = msg_config.get("max_response_length", config.max_response_length) config.max_response_length = msg_config.get("max_response_length", config.max_response_length)
if config.INNER_VERSION in SpecifierSet(">=1.1.4"): if config.INNER_VERSION in SpecifierSet(">=1.1.4"):
@@ -587,6 +588,9 @@ class BotConfig:
keywords_reaction_config = parent["keywords_reaction"] keywords_reaction_config = parent["keywords_reaction"]
if keywords_reaction_config.get("enable", False): if keywords_reaction_config.get("enable", False):
config.keywords_reaction_rules = keywords_reaction_config.get("rules", config.keywords_reaction_rules) config.keywords_reaction_rules = keywords_reaction_config.get("rules", config.keywords_reaction_rules)
for rule in config.keywords_reaction_rules:
if rule.get("enable", False) and "regex" in rule:
rule["regex"] = [re.compile(r) for r in rule.get("regex", [])]
def chinese_typo(parent: dict): def chinese_typo(parent: dict):
chinese_typo_config = parent["chinese_typo"] chinese_typo_config = parent["chinese_typo"]

View File

@@ -149,6 +149,11 @@ enable = false # 仅作示例,不会触发
keywords = ["测试关键词回复","test",""] keywords = ["测试关键词回复","test",""]
reaction = "回答“测试成功”" reaction = "回答“测试成功”"
[[keywords_reaction.rules]] # 使用正则表达式匹配句式
enable = false # 仅作示例,不会触发
regex = ["^(?P<n>\\S{1,20})是这样的$"] # 将匹配到的词汇命名为n反应中对应的[n]会被替换为匹配到的内容,若不了解正则表达式请勿编写
reaction = "请按照以下模板造句:[n]是这样的xx只要xx就可以可是[n]要考虑的事情就很多了比如什么时候xx什么时候xx什么时候xx。请自由发挥替换xx部分只需保持句式结构同时表达一种将[n]过度重视的反讽意味)"
[chinese_typo] [chinese_typo]
enable = true # 是否启用中文错别字生成器 enable = true # 是否启用中文错别字生成器
error_rate=0.001 # 单字替换概率 error_rate=0.001 # 单字替换概率